[Fix-16991] Missing environmentConfig when retry/failover/recover task instance

Author: ruanwenjun

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/IEnvironmentDao.java

@@ -0,0 +1,27 @@
+/*
+ * 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.dao.repository;
+
+import org.apache.dolphinscheduler.dao.entity.Environment;
+
+import java.util.Optional;
+
+public interface IEnvironmentDao extends IDao<Environment> {
+
+    Optional<Environment> queryByEnvironmentCode(Long environmentCode);
+}

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/repository/impl/EnvironmentDaoImpl.java

@@ -0,0 +1,42 @@
+/*
+ * 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.dao.repository.impl;
+
+import org.apache.dolphinscheduler.dao.entity.Environment;
+import org.apache.dolphinscheduler.dao.mapper.EnvironmentMapper;
+import org.apache.dolphinscheduler.dao.repository.BaseDao;
+import org.apache.dolphinscheduler.dao.repository.IEnvironmentDao;
+
+import java.util.Optional;
+
+import lombok.NonNull;
+
+import org.springframework.stereotype.Repository;
+
+@Repository
+public class EnvironmentDaoImpl extends BaseDao<Environment, EnvironmentMapper> implements IEnvironmentDao {
+
+    public EnvironmentDaoImpl(@NonNull EnvironmentMapper environmentMapper) {
+        super(environmentMapper);
+    }
+
+    @Override
+    public Optional<Environment> queryByEnvironmentCode(Long environmentCode) {
+        return Optional.ofNullable(mybatisMapper.queryByEnvironmentCode(environmentCode));
+    }
+}

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/TaskInstanceUtils.java

@@ -26,6 +26,8 @@
 import org.apache.dolphinscheduler.server.master.engine.executor.plugin.ITaskParameterDeserializer;
 import org.apache.dolphinscheduler.server.master.exception.MasterTaskExecuteException;
 
+import org.apache.commons.lang3.StringUtils;
+
 import lombok.extern.slf4j.Slf4j;
 
 import com.fasterxml.jackson.core.type.TypeReference;
@@ -51,9 +53,13 @@ public void start() throws MasterTaskExecuteException {
         try {
             log.info("Begin to execute LogicFakeTask: {}", taskExecutionContext.getTaskName());
 
-            final String shellScript = ParameterUtils.convertParameterPlaceholders(
+            String shellScript = ParameterUtils.convertParameterPlaceholders(
                     taskParameters.getShellScript(),
                     ParameterUtils.convert(taskExecutionContext.getPrepareParamsMap()));
+
+            if (StringUtils.isNotEmpty(taskExecutionContext.getEnvironmentConfig())) {
+                shellScript = taskExecutionContext.getEnvironmentConfig() + "\n" + shellScript;
+            }
             final String[] cmd = {"/bin/sh", "-c", shellScript};
             process = Runtime.getRuntime().exec(cmd);
             int exitCode = process.waitFor();

dolphinscheduler-dao/src/test/java/org/apache/dolphinscheduler/dao/utils/TaskInstanceUtilsTest.java

@@ -17,7 +17,6 @@
 
 package org.apache.dolphinscheduler.server.master.engine.task.runnable;
 
-import org.apache.dolphinscheduler.dao.entity.Environment;
 import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
@@ -62,7 +61,6 @@ protected TaskInstance cloneTaskInstance(TaskInstance originTaskInstance) {
         result.setTaskInstancePriority(originTaskInstance.getTaskInstancePriority());
         result.setWorkerGroup(originTaskInstance.getWorkerGroup());
         result.setEnvironmentCode(originTaskInstance.getEnvironmentCode());
-        result.setEnvironmentConfig(originTaskInstance.getEnvironmentConfig());
         result.setExecutorId(originTaskInstance.getExecutorId());
         result.setVarPool(originTaskInstance.getVarPool());
         result.setExecutorName(originTaskInstance.getExecutorName());
@@ -116,14 +114,4 @@ protected void injectMetadataFromWorkflowInstance(TaskInstance taskInstance, Wor
         taskInstance.setTestFlag(workflowInstance.getTestFlag());
     }
 
-    protected void injectEnvironmentConfigFromDB(TaskInstance taskInstance) {
-        if (EnvironmentUtils.isEnvironmentCodeEmpty(taskInstance.getEnvironmentCode())) {
-            return;
-        }
-        Environment environment = environmentMapper.queryByEnvironmentCode(taskInstance.getEnvironmentCode());
-        if (environment == null) {
-            throw new IllegalArgumentException("Cannot find the environment: " + taskInstance.getEnvironmentCode());
-        }
-        taskInstance.setEnvironmentConfig(environment.getConfig());
-    }
 }

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/executor/plugin/fake/LogicFakeTask.java

@@ -51,7 +51,6 @@ public TaskInstance createTaskInstance(FirstRunTaskInstanceBuilder builder) {
         TaskInstance taskInstance = new TaskInstance();
         injectMetadataFromTaskDefinition(taskInstance, taskDefinition);
         injectMetadataFromWorkflowInstance(taskInstance, workflowInstance);
-        injectEnvironmentConfigFromDB(taskInstance);
 
         taskInstance.setState(TaskExecutionStatus.SUBMITTED_SUCCESS);
         taskInstance.setFirstSubmitTime(new Date());

dolphinscheduler-master/src/main/java/org/apache/dolphinscheduler/server/master/engine/task/runnable/AbstractTaskInstanceFactory.java

@@ -23,11 +23,9 @@
 import org.apache.dolphinscheduler.common.utils.DateUtils;
 import org.apache.dolphinscheduler.dao.entity.TaskDefinition;
 import org.apache.dolphinscheduler.dao.entity.TaskInstance;
-import org.apache.dolphinscheduler.dao.entity.WorkflowDefinition;
 import org.apache.dolphinscheduler.dao.entity.WorkflowInstance;
 import org.apache.dolphinscheduler.plugin.task.api.K8sTaskExecutionContext;
 import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
-import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
 import org.apache.dolphinscheduler.plugin.task.api.enums.TaskTimeoutStrategy;
 import org.apache.dolphinscheduler.plugin.task.api.model.Property;
 import org.apache.dolphinscheduler.plugin.task.api.parameters.resource.ResourceParametersHelper;
@@ -48,35 +46,37 @@ public static TaskExecutionContextBuilder get() {
         return new TaskExecutionContextBuilder();
     }
 
-    private TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
+    private final TaskExecutionContext taskExecutionContext;
+
+    public TaskExecutionContextBuilder() {
+        this.taskExecutionContext = new TaskExecutionContext();
+    }
 
     /**
      * build taskInstance related info
      *
      * @param taskInstance taskInstance
      * @return TaskExecutionContextBuilder
      */
-    public TaskExecutionContextBuilder buildTaskInstanceRelatedInfo(TaskInstance taskInstance) {
+    public TaskExecutionContextBuilder buildTaskInstanceRelatedInfo(final TaskInstance taskInstance) {
         taskExecutionContext.setTaskInstanceId(taskInstance.getId());
         taskExecutionContext.setTaskName(taskInstance.getName());
         taskExecutionContext.setFirstSubmitTime(DateUtils.dateToTimeStamp(taskInstance.getFirstSubmitTime()));
         taskExecutionContext.setStartTime(DateUtils.dateToTimeStamp(taskInstance.getStartTime()));
         taskExecutionContext.setTaskType(taskInstance.getTaskType());
         taskExecutionContext.setLogPath(taskInstance.getLogPath());
         taskExecutionContext.setWorkerGroup(taskInstance.getWorkerGroup());
-        taskExecutionContext.setEnvironmentConfig(taskInstance.getEnvironmentConfig());
         taskExecutionContext.setHost(taskInstance.getHost());
         taskExecutionContext.setVarPool(taskInstance.getVarPool());
         taskExecutionContext.setDryRun(taskInstance.getDryRun());
         taskExecutionContext.setTestFlag(taskInstance.getTestFlag());
-        taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.SUBMITTED_SUCCESS);
         taskExecutionContext.setCpuQuota(taskInstance.getCpuQuota());
         taskExecutionContext.setMemoryMax(taskInstance.getMemoryMax());
         taskExecutionContext.setAppIds(taskInstance.getAppLink());
         return this;
     }
 
-    public TaskExecutionContextBuilder buildTaskDefinitionRelatedInfo(TaskDefinition taskDefinition) {
+    public TaskExecutionContextBuilder buildTaskDefinitionRelatedInfo(final TaskDefinition taskDefinition) {
         // todo: remove the timeout setting here the timeout strategy should be used at master
         taskExecutionContext.setTaskTimeout(Integer.MAX_VALUE);
         if (taskDefinition.getTimeoutFlag() == TimeoutFlag.OPEN) {
@@ -97,7 +97,7 @@ public TaskExecutionContextBuilder buildTaskDefinitionRelatedInfo(TaskDefinition
      * @param workflowInstance processInstance
      * @return TaskExecutionContextBuilder
      */
-    public TaskExecutionContextBuilder buildProcessInstanceRelatedInfo(WorkflowInstance workflowInstance) {
+    public TaskExecutionContextBuilder buildProcessInstanceRelatedInfo(final WorkflowInstance workflowInstance) {
         taskExecutionContext.setWorkflowInstanceId(workflowInstance.getId());
         taskExecutionContext.setScheduleTime(DateUtils.dateToTimeStamp(workflowInstance.getScheduleTime()));
         taskExecutionContext.setGlobalParams(workflowInstance.getGlobalParams());
@@ -110,20 +110,7 @@ public TaskExecutionContextBuilder buildProcessInstanceRelatedInfo(WorkflowInsta
         return this;
     }
 
-    /**
-     * build processDefinition related info
-     *
-     * @param workflowDefinition processDefinition
-     * @return TaskExecutionContextBuilder
-     */
-    public TaskExecutionContextBuilder buildProcessDefinitionRelatedInfo(WorkflowDefinition workflowDefinition) {
-        taskExecutionContext.setWorkflowDefinitionCode(workflowDefinition.getCode());
-        taskExecutionContext.setWorkflowDefinitionVersion(workflowDefinition.getVersion());
-        taskExecutionContext.setProjectCode(workflowDefinition.getProjectCode());
-        return this;
-    }
-
-    public TaskExecutionContextBuilder buildResourceParametersInfo(ResourceParametersHelper parametersHelper) {
+    public TaskExecutionContextBuilder buildResourceParameters(final ResourceParametersHelper parametersHelper) {
         taskExecutionContext.setResourceParametersHelper(parametersHelper);
         return this;
     }
@@ -135,7 +122,7 @@ public TaskExecutionContextBuilder buildResourceParametersInfo(ResourceParameter
      * @return TaskExecutionContextBuilder
      */
 
-    public TaskExecutionContextBuilder buildK8sTaskRelatedInfo(K8sTaskExecutionContext k8sTaskExecutionContext) {
+    public TaskExecutionContextBuilder buildK8sTaskRelatedInfo(final K8sTaskExecutionContext k8sTaskExecutionContext) {
         taskExecutionContext.setK8sTaskExecutionContext(k8sTaskExecutionContext);
         return this;
     }
@@ -146,7 +133,7 @@ public TaskExecutionContextBuilder buildK8sTaskRelatedInfo(K8sTaskExecutionConte
      * @param propertyMap
      * @return
      */
-    public TaskExecutionContextBuilder buildParamInfo(Map<String, Property> propertyMap) {
+    public TaskExecutionContextBuilder buildPrepareParams(final Map<String, Property> propertyMap) {
         taskExecutionContext.setPrepareParamsMap(propertyMap);
         return this;
     }
@@ -157,16 +144,21 @@ public TaskExecutionContextBuilder buildParamInfo(Map<String, Property> property
      * @param businessParamsMap
      * @return
      */
-    public TaskExecutionContextBuilder buildBusinessParamsMap(Map<String, Property> businessParamsMap) {
+    public TaskExecutionContextBuilder buildBusinessParams(final Map<String, Property> businessParamsMap) {
         taskExecutionContext.setParamsMap(businessParamsMap);
         return this;
     }
 
-    public TaskExecutionContextBuilder buildWorkflowInstanceHost(String masterHost) {
+    public TaskExecutionContextBuilder buildWorkflowInstanceHost(final String masterHost) {
         taskExecutionContext.setWorkflowInstanceHost(masterHost);
         return this;
     }
 
+    public TaskExecutionContextBuilder buildEnvironmentConfig(final String environmentConfig) {
+        taskExecutionContext.setEnvironmentConfig(environmentConfig);
+        return this;
+    }
+
     /**
      * create
      *