[Fix-16768] Add scheduleTime parameter to sub-workflow instance.

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/subworkflow/SubWorkflowControlClient.java

@@ -30,11 +30,11 @@
 import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceRecoverSuspendTasksResponse;
 import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
 import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
-import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerRequest;
 import org.apache.dolphinscheduler.server.master.engine.workflow.trigger.WorkflowInstanceRecoverFailureTaskTrigger;
 import org.apache.dolphinscheduler.server.master.engine.workflow.trigger.WorkflowInstanceRecoverSuspendTaskTrigger;
 import org.apache.dolphinscheduler.server.master.exception.MasterTaskExecuteException;
-import org.apache.dolphinscheduler.server.master.runner.task.subworkflow.trigger.SubWorkflowManualTrigger;
+import org.apache.dolphinscheduler.server.master.runner.task.subworkflow.trigger.SubWorkflowTrigger;
+import org.apache.dolphinscheduler.server.master.runner.task.subworkflow.trigger.SubWorkflowTriggerRequest;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -49,16 +49,16 @@ public class SubWorkflowControlClient {
     private WorkflowInstanceDao workflowInstanceDao;
 
     @Autowired
-    private SubWorkflowManualTrigger subWorkflowManualTrigger;
+    private SubWorkflowTrigger subWorkflowTrigger;
 
     @Autowired
     private WorkflowInstanceRecoverFailureTaskTrigger workflowInstanceRecoverFailureTaskTrigger;
 
     @Autowired
     private WorkflowInstanceRecoverSuspendTaskTrigger workflowInstanceRecoverSuspendTaskTrigger;
 
-    public Integer triggerSubWorkflow(final WorkflowManualTriggerRequest workflowManualTriggerRequest) {
-        return subWorkflowManualTrigger.triggerWorkflow(workflowManualTriggerRequest).getWorkflowInstanceId();
+    public Integer triggerSubWorkflow(final SubWorkflowTriggerRequest subWorkflowTriggerRequest) {
+        return subWorkflowTrigger.triggerWorkflow(subWorkflowTriggerRequest).getWorkflowInstanceId();
     }
 
     public WorkflowInstanceRecoverFailureTasksResponse triggerFromFailureTasks(

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/subworkflow/SubWorkflowLogicTask.java

@@ -32,14 +32,14 @@
 import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceRecoverSuspendTasksRequest;
 import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopRequest;
 import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowInstanceStopResponse;
-import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerRequest;
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.SubWorkflowParameters;
 import org.apache.dolphinscheduler.server.master.engine.workflow.runnable.IWorkflowExecutionRunnable;
 import org.apache.dolphinscheduler.server.master.exception.MasterTaskExecuteException;
 import org.apache.dolphinscheduler.server.master.runner.execute.AsyncTaskExecuteFunction;
 import org.apache.dolphinscheduler.server.master.runner.message.LogicTaskInstanceExecutionEventSenderManager;
 import org.apache.dolphinscheduler.server.master.runner.task.BaseAsyncLogicTask;
+import org.apache.dolphinscheduler.server.master.runner.task.subworkflow.trigger.SubWorkflowTriggerRequest;
 
 import lombok.extern.slf4j.Slf4j;
 
@@ -183,7 +183,8 @@ private SubWorkflowLogicTaskRuntimeContext triggerNewSubWorkflow() {
         final ICommandParam commandParam =
                 JSONUtils.parseObject(workflowInstance.getCommandParam(), ICommandParam.class);
 
-        final WorkflowManualTriggerRequest workflowManualTriggerRequest = WorkflowManualTriggerRequest.builder()
+        final SubWorkflowTriggerRequest workflowManualTriggerRequest = SubWorkflowTriggerRequest.builder()
+                .scheduleTIme(workflowInstance.getScheduleTime())
                 .userId(taskExecutionContext.getExecutorId())
                 .workflowDefinitionCode(subWorkflowDefinition.getCode())
                 .workflowDefinitionVersion(subWorkflowDefinition.getVersion())

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/subworkflow/trigger/SubWorkflowTrigger.java

@@ -0,0 +1,111 @@
+/*
+ * 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.runner.task.subworkflow.trigger;
+
+import org.apache.dolphinscheduler.common.enums.CommandType;
+import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.WarningType;
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.common.utils.DateUtils;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.dao.entity.Command;
+import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
+import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import org.apache.dolphinscheduler.dao.utils.EnvironmentUtils;
+import org.apache.dolphinscheduler.dao.utils.WorkerGroupUtils;
+import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
+import org.apache.dolphinscheduler.server.master.engine.workflow.trigger.AbstractWorkflowTrigger;
+
+import org.apache.commons.lang3.ObjectUtils;
+
+import java.util.Date;
+
+import org.springframework.stereotype.Component;
+
+/**
+ * Used to trigger the sub-workflow and generate the workflow instance.
+ */
+@Component
+public class SubWorkflowTrigger
+        extends
+            AbstractWorkflowTrigger<SubWorkflowTriggerRequest, SubWorkflowTriggerResponse> {
+
+    @Override
+    protected WorkflowInstance constructWorkflowInstance(final SubWorkflowTriggerRequest subWorkflowTriggerRequest) {
+        final CommandType commandType = CommandType.START_PROCESS;
+        final Long workflowCode = subWorkflowTriggerRequest.getWorkflowDefinitionCode();
+        final Integer workflowVersion = subWorkflowTriggerRequest.getWorkflowDefinitionVersion();
+        final WorkflowDefinition workflowDefinition = getProcessDefinition(workflowCode, workflowVersion);
+
+        final WorkflowInstance workflowInstance = new WorkflowInstance();
+        workflowInstance.setWorkflowDefinitionCode(workflowDefinition.getCode());
+        workflowInstance.setWorkflowDefinitionVersion(workflowDefinition.getVersion());
+        workflowInstance.setProjectCode(workflowDefinition.getProjectCode());
+        workflowInstance.setCommandType(commandType);
+        workflowInstance.setStateWithDesc(WorkflowExecutionStatus.SUBMITTED_SUCCESS, commandType.name());
+        workflowInstance.setRecovery(Flag.NO);
+        workflowInstance.setScheduleTime(subWorkflowTriggerRequest.getScheduleTIme());
+        workflowInstance.setStartTime(new Date());
+        workflowInstance.setRestartTime(workflowInstance.getStartTime());
+        workflowInstance.setRunTimes(1);
+        workflowInstance.setName(String.join("-", workflowDefinition.getName(), DateUtils.getCurrentTimeStamp()));
+        workflowInstance.setTaskDependType(subWorkflowTriggerRequest.getTaskDependType());
+        workflowInstance.setFailureStrategy(subWorkflowTriggerRequest.getFailureStrategy());
+        workflowInstance.setWarningType(
+                ObjectUtils.defaultIfNull(subWorkflowTriggerRequest.getWarningType(), WarningType.NONE));
+        workflowInstance.setWarningGroupId(subWorkflowTriggerRequest.getWarningGroupId());
+        workflowInstance.setExecutorId(subWorkflowTriggerRequest.getUserId());
+        workflowInstance.setExecutorName(getExecutorUser(subWorkflowTriggerRequest.getUserId()).getUserName());
+        workflowInstance.setTenantCode(subWorkflowTriggerRequest.getTenantCode());
+        workflowInstance.setIsSubWorkflow(Flag.YES);
+        workflowInstance.addHistoryCmd(commandType);
+        workflowInstance.setWorkflowInstancePriority(subWorkflowTriggerRequest.getWorkflowInstancePriority());
+        workflowInstance.setWorkerGroup(
+                WorkerGroupUtils.getWorkerGroupOrDefault(subWorkflowTriggerRequest.getWorkerGroup()));
+        workflowInstance.setEnvironmentCode(
+                EnvironmentUtils.getEnvironmentCodeOrDefault(subWorkflowTriggerRequest.getEnvironmentCode()));
+        workflowInstance.setTimeout(workflowDefinition.getTimeout());
+        workflowInstance.setDryRun(subWorkflowTriggerRequest.getDryRun().getCode());
+        workflowInstance.setTestFlag(subWorkflowTriggerRequest.getTestFlag().getCode());
+        return workflowInstance;
+    }
+
+    @Override
+    protected Command constructTriggerCommand(final SubWorkflowTriggerRequest subWorkflowTriggerRequest,
+                                              final WorkflowInstance workflowInstance) {
+        final RunWorkflowCommandParam runWorkflowCommandParam = RunWorkflowCommandParam.builder()
+                .commandParams(subWorkflowTriggerRequest.getStartParamList())
+                .startNodes(subWorkflowTriggerRequest.getStartNodes())
+                .timeZone(DateUtils.getTimezone())
+                .build();
+        return Command.builder()
+                .commandType(CommandType.START_PROCESS)
+                .workflowDefinitionCode(subWorkflowTriggerRequest.getWorkflowDefinitionCode())
+                .workflowDefinitionVersion(subWorkflowTriggerRequest.getWorkflowDefinitionVersion())
+                .workflowInstanceId(workflowInstance.getId())
+                .workflowInstancePriority(workflowInstance.getWorkflowInstancePriority())
+                .commandParam(JSONUtils.toJsonString(runWorkflowCommandParam))
+                .build();
+    }
+
+    @Override
+    protected SubWorkflowTriggerResponse onTriggerSuccess(WorkflowInstance workflowInstance) {
+        return SubWorkflowTriggerResponse.success(workflowInstance.getId());
+    }
+
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/subworkflow/trigger/SubWorkflowTriggerRequest.java

@@ -0,0 +1,81 @@
+/*
+ * 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.runner.task.subworkflow.trigger;
+
+import org.apache.dolphinscheduler.common.enums.FailureStrategy;
+import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.Priority;
+import org.apache.dolphinscheduler.common.enums.TaskDependType;
+import org.apache.dolphinscheduler.common.enums.WarningType;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+
+import java.util.ArrayList;
+import java.util.Date;
+import java.util.List;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SubWorkflowTriggerRequest {
+
+    private Date scheduleTIme;
+
+    private Integer userId;
+
+    private Long workflowDefinitionCode;
+
+    private Integer workflowDefinitionVersion;
+
+    private List<Long> startNodes;
+
+    @Builder.Default
+    private FailureStrategy failureStrategy = FailureStrategy.CONTINUE;
+
+    @Builder.Default
+    private TaskDependType taskDependType = TaskDependType.TASK_POST;
+
+    @Builder.Default
+    private WarningType warningType = WarningType.NONE;
+
+    private Integer warningGroupId;
+
+    @Builder.Default
+    private Priority workflowInstancePriority = Priority.MEDIUM;
+
+    private String workerGroup;
+
+    private String tenantCode;
+
+    private Long environmentCode;
+
+    @Builder.Default
+    private List<Property> startParamList = new ArrayList<>();
+
+    @Builder.Default
+    private Flag dryRun = Flag.NO;
+
+    @Builder.Default
+    private Flag testFlag = Flag.NO;
+
+}

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/runner/task/subworkflow/trigger/SubWorkflowTriggerResponse.java

@@ -0,0 +1,51 @@
+/*
+ * 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.runner.task.subworkflow.trigger;
+
+import lombok.AllArgsConstructor;
+import lombok.Builder;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+@Data
+@Builder
+@NoArgsConstructor
+@AllArgsConstructor
+public class SubWorkflowTriggerResponse {
+
+    private boolean success;
+
+    private String message;
+
+    private Integer workflowInstanceId;
+
+    public static SubWorkflowTriggerResponse fail(String message) {
+        return SubWorkflowTriggerResponse.builder()
+                .success(false)
+                .message(message)
+                .build();
+    }
+
+    public static SubWorkflowTriggerResponse success(Integer workflowInstanceId) {
+        return SubWorkflowTriggerResponse.builder()
+                .success(true)
+                .workflowInstanceId(workflowInstanceId)
+                .build();
+    }
+
+}