[Fix-17534][Service/Master] Add global parameters and varpool from current workflow instance and add them to start params list of the trigger request of a sub workflow

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/subworkflow/SubWorkflowLogicTask.java

@@ -17,6 +17,9 @@
 
 package org.apache.dolphinscheduler.server.master.engine.executor.plugin.subworkflow;
 
+import static java.util.Arrays.asList;
+import static org.apache.dolphinscheduler.plugin.task.api.utils.VarPoolUtils.deserializeVarPool;
+
 import org.apache.dolphinscheduler.common.enums.Flag;
 import org.apache.dolphinscheduler.common.utils.JSONUtils;
 import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
@@ -35,6 +38,7 @@
 import org.apache.dolphinscheduler.extract.master.transportor.workflow.WorkflowManualTriggerRequest;
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.SubWorkflowParameters;
 import org.apache.dolphinscheduler.server.master.engine.executor.plugin.AbstractLogicTask;
 import org.apache.dolphinscheduler.server.master.engine.executor.plugin.ITaskParameterDeserializer;
@@ -43,6 +47,14 @@
 import org.apache.dolphinscheduler.task.executor.ITaskExecutor;
 import org.apache.dolphinscheduler.task.executor.events.TaskExecutorRuntimeContextChangedLifecycleEvent;
 
+import org.apache.commons.collections4.CollectionUtils;
+
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
 import lombok.extern.slf4j.Slf4j;
 
 import org.springframework.context.ApplicationContext;
@@ -222,6 +234,11 @@ private SubWorkflowLogicTaskRuntimeContext triggerNewSubWorkflow() {
         final ICommandParam commandParam =
                 JSONUtils.parseObject(workflowInstance.getCommandParam(), ICommandParam.class);
 
+        final List<Property> paramList = mergeParams(asList(
+                commandParam.getCommandParams(),
+                new ArrayList<>(deserializeVarPool(workflowInstance.getVarPool())),
+                new ArrayList<>(deserializeVarPool(workflowInstance.getGlobalParams()))));
+
         final WorkflowManualTriggerRequest workflowManualTriggerRequest = WorkflowManualTriggerRequest.builder()
                 .userId(taskExecutionContext.getExecutorId())
                 .workflowDefinitionCode(subWorkflowDefinition.getCode())
@@ -233,8 +250,7 @@ private SubWorkflowLogicTaskRuntimeContext triggerNewSubWorkflow() {
                 .workerGroup(workflowInstance.getWorkerGroup())
                 .tenantCode(workflowInstance.getTenantCode())
                 .environmentCode(workflowInstance.getEnvironmentCode())
-                // todo: transport varpool and local params
-                .startParamList(commandParam.getCommandParams())
+                .startParamList(paramList)
                 .dryRun(Flag.of(workflowInstance.getDryRun()))
                 .build();
         final Integer subWorkflowInstanceId = applicationContext
@@ -243,6 +259,25 @@ private SubWorkflowLogicTaskRuntimeContext triggerNewSubWorkflow() {
         return SubWorkflowLogicTaskRuntimeContext.of(subWorkflowInstanceId);
     }
 
+    private List<Property> mergeParams(List<List<Property>> params) {
+        if (CollectionUtils.isEmpty(params)) {
+            return Collections.emptyList();
+        }
+        if (params.size() == 1) {
+            return params.get(0);
+        }
+        Map<String, Property> result = new HashMap<>();
+        for (List<Property> param : params) {
+            if (CollectionUtils.isEmpty(param)) {
+                continue;
+            }
+            for (Property property : param) {
+                result.put(property.getProp(), property);
+            }
+        }
+        return new ArrayList<>(result.values());
+    }
+
     private void upsertSubWorkflowRelation() {
         final WorkflowInstanceMapDao workflowInstanceMapDao = applicationContext.getBean(WorkflowInstanceMapDao.class);
         WorkflowInstanceRelation workflowInstanceRelation = workflowInstanceMapDao.queryWorkflowMapByParent(

dolphinscheduler-master/src/test/java/org/apache/dolphinscheduler/server/master/integration/cases/SubWorkflowInstanceGlobalParamsInheritanceTestCase.java

@@ -0,0 +1,122 @@
+/*
+ * 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.integration.cases;
+
+import static com.google.common.truth.Truth.assertThat;
+import static org.awaitility.Awaitility.await;
+
+import org.apache.dolphinscheduler.common.enums.Flag;
+import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
+import org.apache.dolphinscheduler.common.utils.JSONUtils;
+import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
+import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
+import org.apache.dolphinscheduler.extract.master.command.RunWorkflowCommandParam;
+import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
+import org.apache.dolphinscheduler.plugin.task.api.model.Property;
+import org.apache.dolphinscheduler.server.master.AbstractMasterIntegrationTestCase;
+import org.apache.dolphinscheduler.server.master.integration.WorkflowOperator;
+import org.apache.dolphinscheduler.server.master.integration.WorkflowTestCaseContext;
+
+import java.time.Duration;
+import java.util.List;
+
+import org.assertj.core.api.Assertions;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+/**
+ * The integration test for validating sub workflow instances inherit global params from all parent flows.
+ * Global params are asserted in each sub workflow instance and the fake_task in the sub-sub-workflow is used to verify the global params.
+ */
+class SubWorkflowInstanceGlobalParamsInheritanceTestCase extends AbstractMasterIntegrationTestCase {
+
+    @Test
+    @DisplayName("Test subflows inherit global params from all parent flows")
+    void testSubflowInheritsGlobalParamsFromParentFlows_with_oneSuccessTask() {
+        final String yaml = "/it/start/workflow_with_sub_workflows_with_global_params.yaml";
+        final WorkflowTestCaseContext context = workflowTestCaseContextFactory.initializeContextFromYaml(yaml);
+        final WorkflowDefinition parentWorkflow = context.getOneWorkflow();
+
+        final WorkflowOperator.WorkflowTriggerDTO workflowTriggerDTO = WorkflowOperator.WorkflowTriggerDTO.builder()
+                .workflowDefinition(parentWorkflow)
+                .runWorkflowCommandParam(new RunWorkflowCommandParam())
+                .build();
+        final Integer workflowInstanceId = workflowOperator.manualTriggerWorkflow(workflowTriggerDTO);
+
+        await()
+                .atMost(Duration.ofMinutes(1))
+                .untilAsserted(() -> {
+                    Assertions
+                            .assertThat(repository.queryWorkflowInstance(workflowInstanceId))
+                            .matches(workflowInstance -> workflowInstance.getState() == WorkflowExecutionStatus.SUCCESS)
+                            .matches(workflowInstance -> workflowInstance.getIsSubWorkflow() == Flag.NO);
+
+                    final List<WorkflowInstance> subWorkflowInstance =
+                            repository.queryWorkflowInstance(context.getWorkflows().get(1));
+                    Assertions
+                            .assertThat(subWorkflowInstance)
+                            .hasSize(1)
+                            .satisfiesExactly(workflowInstance -> {
+                                assertThat(workflowInstance.getState()).isEqualTo(WorkflowExecutionStatus.SUCCESS);
+                                assertThat(workflowInstance.getIsSubWorkflow()).isEqualTo(Flag.YES);
+
+                                Assertions
+                                        .assertThat(
+                                                JSONUtils.toList(workflowInstance.getGlobalParams(), Property.class))
+                                        .hasSize(1)
+                                        .anySatisfy(property -> {
+                                            assertThat(property.getProp()).isEqualTo("parentWorkflowParam");
+                                            assertThat(property.getValue()).isEqualTo("parentWorkflowParamValue");
+                                        });
+                            });
+
+                    final List<WorkflowInstance> subSubWorkflowInstance =
+                            repository.queryWorkflowInstance(context.getWorkflows().get(2));
+                    Assertions
+                            .assertThat(subSubWorkflowInstance)
+                            .hasSize(1)
+                            .satisfiesExactly(workflowInstance -> {
+                                assertThat(workflowInstance.getState()).isEqualTo(WorkflowExecutionStatus.SUCCESS);
+                                assertThat(workflowInstance.getIsSubWorkflow()).isEqualTo(Flag.YES);
+
+                                Assertions
+                                        .assertThat(
+                                                JSONUtils.toList(workflowInstance.getGlobalParams(), Property.class))
+                                        .hasSize(2)
+                                        .anySatisfy(property -> {
+                                            assertThat(property.getProp()).isEqualTo("parentWorkflowParam");
+                                            assertThat(property.getValue()).isEqualTo("parentWorkflowParamValue");
+                                        })
+                                        .anySatisfy(property -> {
+                                            assertThat(property.getProp()).isEqualTo("subWorkflowParam");
+                                            assertThat(property.getValue()).isEqualTo("subWorkflowParamValue");
+                                        });
+                            });
+
+                    Assertions
+                            .assertThat(repository.queryTaskInstance(subSubWorkflowInstance.get(0).getId()))
+                            .satisfiesExactly(taskInstance -> {
+                                assertThat(taskInstance.getName()).isEqualTo("fake_task");
+                                assertThat(taskInstance.getState()).isEqualTo(TaskExecutionStatus.SUCCESS);
+                            });
+                });
+
+        masterContainer.assertAllResourceReleased();
+    }
+
+}

dolphinscheduler-master/src/test/resources/it/start/workflow_with_sub_workflows_with_global_params.yaml

@@ -0,0 +1,135 @@
+#
+# 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.
+#
+
+project:
+  name: MasterIntegrationTest
+  code: 1
+  description: This is a fake project
+  userId: 1
+  userName: admin
+  createTime: 2024-08-12 00:00:00
+  updateTime: 2021-08-12 00:00:00
+
+workflows:
+  - name: workflow_with_sub_workflow
+    code: 1
+    version: 1
+    projectCode: 1
+    description: This is a fake workflow with single subflow task
+    releaseState: ONLINE
+    createTime: 2024-08-12 00:00:00
+    updateTime: 2021-08-12 00:00:00
+    userId: 1
+    globalParams: >
+      [{
+        "prop": "parentWorkflowParam",
+        "value": "parentWorkflowParamValue",
+        "direct": "IN",
+        "type":"VARCHAR"
+      }]
+    executionType: PARALLEL
+  - name: subworkflow_with_another_subworkflow
+    code: 2
+    version: 1
+    projectCode: 1
+    description: This is a fake sub workflow with another single subflow task
+    releaseState: ONLINE
+    createTime: 2024-08-12 00:00:00
+    updateTime: 2021-08-12 00:00:00
+    userId: 1
+    executionType: PARALLEL
+  - name: subworkflow_with_single_task
+    code: 3
+    version: 1
+    projectCode: 1
+    description: This is a fake sub sub workflow with single task
+    releaseState: ONLINE
+    createTime: 2024-08-12 00:00:00
+    updateTime: 2021-08-12 00:00:00
+    userId: 1
+    globalParams: >
+      [{
+        "prop": "subWorkflowParam",
+        "value": "subWorkflowParamValue",
+        "direct": "IN",
+        "type":"VARCHAR"
+      }]
+    executionType: PARALLEL
+
+tasks:
+  - name: sub_workflow_task
+    code: 1
+    version: 1
+    projectCode: 1
+    userId: 1
+    taskType: SUB_WORKFLOW
+    taskParams: '{"localParams":[],"resourceList":[],"workflowDefinitionCode":2}'
+    workerGroup: default
+    createTime: 2024-08-12 00:00:00
+    updateTime: 2021-08-12 00:00:00
+    taskExecuteType: BATCH
+  - name: sub_sub_workflow_task
+    code: 2
+    version: 1
+    projectCode: 1
+    userId: 1
+    taskType: SUB_WORKFLOW
+    taskParams: '{"localParams":[],"resourceList":[],"workflowDefinitionCode":3}'
+    workerGroup: default
+    createTime: 2024-08-12 00:00:00
+    updateTime: 2021-08-12 00:00:00
+    taskExecuteType: BATCH
+  - name: fake_task
+    code: 3
+    version: 1
+    projectCode: 1
+    userId: 1
+    taskType: LogicFakeTask
+    taskParams: '{"localParams":null,"varPool":[],"shellScript":  "if [ \"${parentWorkflowParam}\" != \"parentWorkflowParamValue\" ]; then\n  exit 1\nelif [ \"${subWorkflowParam}\" != \"subWorkflowParamValue\" ]; then\n  exit 1\nelse\n  exit 0\nfi"}'
+    workerGroup: default
+    createTime: 2024-08-12 00:00:00
+    updateTime: 2021-08-12 00:00:00
+    taskExecuteType: BATCH
+
+taskRelations:
+  - projectCode: 1
+    workflowDefinitionCode: 1
+    workflowDefinitionVersion: 1
+    preTaskCode: 0
+    preTaskVersion: 0
+    postTaskCode: 1
+    postTaskVersion: 1
+    createTime: 2024-08-12 00:00:00
+    updateTime: 2024-08-12 00:00:00
+  - projectCode: 1
+    workflowDefinitionCode: 2
+    workflowDefinitionVersion: 1
+    preTaskCode: 0
+    preTaskVersion: 0
+    postTaskCode: 2
+    postTaskVersion: 1
+    createTime: 2024-08-12 00:00:00
+    updateTime: 2024-08-12 00:00:00
+  - projectCode: 1
+    workflowDefinitionCode: 3
+    workflowDefinitionVersion: 1
+    preTaskCode: 0
+    preTaskVersion: 0
+    postTaskCode: 3
+    postTaskVersion: 1
+    createTime: 2024-08-12 00:00:00
+    updateTime: 2024-08-12 00:00:00