natas_hw
/
dolphinscheduler
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'upstream/dev' into spilit 

# Conflicts:
#	dolphinscheduler-common/src/main/resources/common.properties
This commit is contained in:
lenboo 2021-03-30 11:31:42 +08:00
parent 689106d114 9c3cec5bbd
commit c8ac6f99fe
21 changed files with 170 additions and 86 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -87,7 +87,7 @@ We would like to express our deep gratitude to all the open-source projects used
## Community
You are very welcome to communicate with the developers and users of Dolphin Scheduler. There are two ways to find them:
1. Join the Slack channel by [this invitation link](https://join.slack.com/t/asf-dolphinscheduler/shared_invite/zt-mzqu52gi-rCggPkSHQ0DZYkwbTxO1Gw).
1. Join the Slack channel by [this invitation link](https://join.slack.com/t/asf-dolphinscheduler/shared_invite/zt-omtdhuio-_JISsxYhiVsltmC5h38yfw).
2. Follow the [Twitter account of Dolphin Scheduler](https://twitter.com/dolphinschedule) and get the latest news on time.
## How to Contribute

2
README_zh_CN.md
View File

@ -92,7 +92,7 @@ Dolphin Scheduler使用了很多优秀的开源项目，比如google的guava、g
## 社区
1. 通过[该申请链接](https://join.slack.com/t/asf-dolphinscheduler/shared_invite/zt-mzqu52gi-rCggPkSHQ0DZYkwbTxO1Gw)加入slack channel
1. 通过[该申请链接](https://join.slack.com/t/asf-dolphinscheduler/shared_invite/zt-omtdhuio-_JISsxYhiVsltmC5h38yfw)加入slack channel
2. 关注[Apache Dolphin Scheduler的Twitter账号](https://twitter.com/dolphinschedule)获取实时动态
## 版权

5
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
View File

@ -155,6 +155,11 @@ public final class Constants {
public static final String DEVELOPMENT_STATE = "development.state";
public static final String DEVELOPMENT_STATE_DEFAULT_VALUE = "true";
/**
* sudo enable
*/
public static final String SUDO_ENABLE = "sudo.enable";
/**
* string true
*/

7
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/CommonUtils.java
View File

@ -68,6 +68,13 @@ public class CommonUtils {
return PropertyUtils.getBoolean(Constants.DEVELOPMENT_STATE, true);
}
/**
* @return sudo enable
*/
public static boolean isSudoEnable() {
return PropertyUtils.getBoolean(Constants.SUDO_ENABLE, true);
}
/**
* if upload resource is HDFS and kerberos startup is true , else false
*

23
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/FileUtils.java
View File

@ -121,27 +121,8 @@ public class FileUtils {
* @return directory of process execution
*/
public static String getProcessExecDir(int projectId, int processDefineId, int processInstanceId, int taskInstanceId) {
String fileName = String.format("%s/exec/process/%s/%s/%s/%s", DATA_BASEDIR, Integer.toString(projectId),
Integer.toString(processDefineId), Integer.toString(processInstanceId), Integer.toString(taskInstanceId));
File file = new File(fileName);
if (!file.getParentFile().exists()) {
file.getParentFile().mkdirs();
}
return fileName;
}
/**
* directory of process instances
*
* @param projectId project id
* @param processDefineId process definition id
* @param processInstanceId process instance id
* @return directory of process instances
*/
public static String getProcessExecDir(int projectId, int processDefineId, int processInstanceId) {
String fileName = String.format("%s/exec/process/%s/%s/%s", DATA_BASEDIR, Integer.toString(projectId),
Integer.toString(processDefineId), Integer.toString(processInstanceId));
String fileName = String.format("%s/exec/process/%d/%d/%d/%d", DATA_BASEDIR,
projectId, processDefineId, processInstanceId, taskInstanceId);
File file = new File(fileName);
if (!file.getParentFile().exists()) {
file.getParentFile().mkdirs();

10
dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/utils/OSUtils.java
View File

@ -22,12 +22,12 @@ import org.apache.dolphinscheduler.common.shell.ShellExecutor;
import org.apache.commons.configuration.Configuration;
import java.lang.management.OperatingSystemMXBean;
import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.lang.management.ManagementFactory;
import java.lang.management.OperatingSystemMXBean;
import java.lang.management.RuntimeMXBean;
import java.math.RoundingMode;
import java.text.DecimalFormat;
@ -406,13 +406,17 @@ public class OSUtils {
}
/**
* get sudo command
* get sudo command
*
* @param tenantCode tenantCode
* @param command command
* @return result of sudo execute command
*/
public static String getSudoCmd(String tenantCode, String command) {
return StringUtils.isEmpty(tenantCode) ? command : "sudo -u " + tenantCode + " " + command;
if (!CommonUtils.isSudoEnable() || StringUtils.isEmpty(tenantCode)) {
return command;
}
return String.format("sudo -u %s %s", tenantCode, command);
}
/**

3
dolphinscheduler-common/src/main/resources/common.properties
View File

@ -78,3 +78,6 @@ datasource.encryption.salt=!@#$%^&*
#dolphin.scheduler.network.priority.strategy=default
# 0<dataCenterId<32
#data.center.id=1
# use sudo or not, if set true ,executing user is tenant user and deploy user need sudo permissions ; if set false, executing user is the deploy user, don't need sudo permissions.
sudo.enable=true

4
dolphinscheduler-common/src/test/java/org/apache/dolphinscheduler/common/utils/FileUtilsTest.java
View File

@ -60,10 +60,8 @@ public class FileUtilsTest {
@Test
public void testGetProcessExecDir() {
String dir = FileUtils.getProcessExecDir(1,2,3, 4);
String dir = FileUtils.getProcessExecDir(1, 2, 3, 4);
Assert.assertEquals("/tmp/dolphinscheduler/exec/process/1/2/3/4", dir);
dir = FileUtils.getProcessExecDir(1,2,3);
Assert.assertEquals("/tmp/dolphinscheduler/exec/process/1/2/3", dir);
}
@Test

2
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/MasterServer.java
View File

@ -43,10 +43,12 @@ import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.annotation.FilterType;
import org.springframework.transaction.annotation.EnableTransactionManagement;
@ComponentScan(value = "org.apache.dolphinscheduler", excludeFilters = {
@ComponentScan.Filter(type = FilterType.ASSIGNABLE_TYPE, classes = {WorkerServer.class})
})
@EnableTransactionManagement
public class MasterServer implements IStoppable {
/**

2
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/config/MasterConfig.java
View File

@ -37,7 +37,7 @@ public class MasterConfig {
@Value("${master.task.commit.retryTimes:5}")
private int masterTaskCommitRetryTimes;
@Value("${master.dispatch.task.num :3}")
@Value("${master.dispatch.task.num:3}")
private int masterDispatchTaskNumber;
@Value("${master.task.commit.interval:1000}")

33
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/master/runner/MasterExecThread.java
View File

@ -42,7 +42,6 @@ import org.apache.dolphinscheduler.common.process.Property;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.common.utils.CollectionUtils;
import org.apache.dolphinscheduler.common.utils.CommonUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
@ -61,16 +60,11 @@ import org.apache.dolphinscheduler.service.process.ProcessService;
import org.apache.dolphinscheduler.service.quartz.cron.CronUtils;
import org.apache.dolphinscheduler.service.queue.PeerTaskInstancePriorityQueue;
import org.apache.commons.io.FileUtils;
import java.io.File;
import java.io.IOException;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
@ -235,8 +229,6 @@ public class MasterExecThread implements Runnable {
processService.updateProcessInstance(processInstance);
} finally {
taskExecService.shutdown();
// post handle
postHandle();
}
}
@ -431,27 +423,6 @@ public class MasterExecThread implements Runnable {
}
}
/**
* process post handle
*/
private void postHandle() {
logger.info("develop mode is: {}", CommonUtils.isDevelopMode());
if (!CommonUtils.isDevelopMode()) {
// get exec dir
String execLocalPath = org.apache.dolphinscheduler.common.utils.FileUtils
.getProcessExecDir(processInstance.getProcessDefinition().getProjectId(),
processInstance.getProcessDefinitionId(),
processInstance.getId());
try {
FileUtils.deleteDirectory(new File(execLocalPath));
} catch (IOException e) {
logger.error("delete exec dir failed ", e);
}
}
}
/**
* submit task to execute
*
@ -921,7 +892,7 @@ public class MasterExecThread implements Runnable {
try {
readyToSubmitTaskQueue.put(taskInstance);
} catch (Exception e) {
logger.error("add task instance to readyToSubmitTaskQueue error");
logger.error("add task instance to readyToSubmitTaskQueue error, taskName: {}", taskInstance.getName(), e);
}
}
@ -935,7 +906,7 @@ public class MasterExecThread implements Runnable {
try {
readyToSubmitTaskQueue.remove(taskInstance);
} catch (Exception e) {
logger.error("remove task instance from readyToSubmitTaskQueue error");
logger.error("remove task instance from readyToSubmitTaskQueue error, taskName: {}", taskInstance.getName(), e);
}
}

4
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/WorkerServer.java
View File

@ -41,17 +41,17 @@ import javax.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.transaction.annotation.EnableTransactionManagement;
/**
* worker server
*/
@ComponentScan("org.apache.dolphinscheduler")
@EnableTransactionManagement
public class WorkerServer implements IStoppable {
/**

5
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/processor/TaskExecuteProcessor.java
View File

@ -20,6 +20,7 @@ package org.apache.dolphinscheduler.server.worker.processor;
import org.apache.dolphinscheduler.common.enums.Event;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.enums.TaskType;
import org.apache.dolphinscheduler.common.utils.CommonUtils;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
@ -141,13 +142,13 @@ public class TaskExecuteProcessor implements NettyRequestProcessor {
// local execute path
String execLocalPath = getExecLocalPath(taskExecutionContext);
logger.info("task instance local execute path : {} ", execLocalPath);
logger.info("task instance local execute path : {}", execLocalPath);
taskExecutionContext.setExecutePath(execLocalPath);
FileUtils.taskLoggerThreadLocal.set(taskLogger);
try {
FileUtils.createWorkDirIfAbsent(execLocalPath);
if (workerConfig.getWorkerTenantAutoCreate()) {
if (CommonUtils.isSudoEnable() && workerConfig.getWorkerTenantAutoCreate()) {
OSUtils.createUserIfAbsent(taskExecutionContext.getTenantCode());
}
} catch (Throwable ex) {

45
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/runner/TaskExecuteThread.java
View File

@ -29,6 +29,7 @@ import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
import org.apache.dolphinscheduler.common.utils.RetryerUtils;
import org.apache.dolphinscheduler.common.utils.StringUtils;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.TaskExecuteAckCommand;
import org.apache.dolphinscheduler.remote.command.TaskExecuteResponseCommand;
@ -45,6 +46,7 @@ import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import org.apache.commons.collections.MapUtils;
import java.io.File;
import java.io.IOException;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
@ -71,17 +73,17 @@ public class TaskExecuteThread implements Runnable, Delayed {
private final Logger logger = LoggerFactory.getLogger(TaskExecuteThread.class);
/**
* task instance
* task instance
*/
private TaskExecutionContext taskExecutionContext;
/**
* abstract task
* abstract task
*/
private AbstractTask task;
/**
* task callback service
* task callback service
*/
private TaskCallbackService taskCallbackService;
@ -185,9 +187,38 @@ public class TaskExecuteThread implements Runnable, Delayed {
responseCommand.setAppIds(task.getAppIds());
} finally {
taskExecutionContextCacheManager.removeByTaskInstanceId(taskExecutionContext.getTaskInstanceId());
ResponceCache.get().cache(taskExecutionContext.getTaskInstanceId(),responseCommand.convert2Command(),Event.RESULT);
ResponceCache.get().cache(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command(), Event.RESULT);
taskCallbackService.sendResult(taskExecutionContext.getTaskInstanceId(), responseCommand.convert2Command());
clearTaskExecPath();
}
}
/**
* when task finish, clear execute path.
*/
private void clearTaskExecPath() {
logger.info("develop mode is: {}", CommonUtils.isDevelopMode());
if (!CommonUtils.isDevelopMode()) {
// get exec dir
String execLocalPath = taskExecutionContext.getExecutePath();
if (StringUtils.isEmpty(execLocalPath)) {
logger.warn("task: {} exec local path is empty.", taskExecutionContext.getTaskName());
return;
}
if ("/".equals(execLocalPath)) {
logger.warn("task: {} exec local path is '/', direct deletion is not allowed", taskExecutionContext.getTaskName());
return;
}
try {
org.apache.commons.io.FileUtils.deleteDirectory(new File(execLocalPath));
logger.info("exec local path: {} cleared.", execLocalPath);
} catch (IOException e) {
logger.error("delete exec dir failed : {}", e.getMessage(), e);
}
}
}
@ -196,7 +227,7 @@ public class TaskExecuteThread implements Runnable, Delayed {
* @return
*/
private Map<String, String> getGlobalParamsMap() {
Map<String,String> globalParamsMap = new HashMap<>(16);
Map<String, String> globalParamsMap = new HashMap<>(16);
// global params string
String globalParamsStr = taskExecutionContext.getGlobalParams();
@ -241,7 +272,7 @@ public class TaskExecuteThread implements Runnable, Delayed {
}
/**
* kill task
* kill task
*/
public void kill() {
if (task != null) {
@ -261,7 +292,7 @@ public class TaskExecuteThread implements Runnable, Delayed {
* @param logger
*/
private void downloadResource(String execLocalPath,
Map<String,String> projectRes,
Map<String, String> projectRes,
Logger logger) throws Exception {
if (MapUtils.isEmpty(projectRes)) {
return;

15
dolphinscheduler-server/src/main/java/org/apache/dolphinscheduler/server/worker/task/AbstractCommandExecutor.java
View File

@ -25,6 +25,7 @@ import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.ExecutionStatus;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.common.utils.CommonUtils;
import org.apache.dolphinscheduler.common.utils.HadoopUtils;
import org.apache.dolphinscheduler.common.utils.LoggerUtils;
import org.apache.dolphinscheduler.common.utils.OSUtils;
@ -84,7 +85,7 @@ public abstract class AbstractCommandExecutor {
* log list
*/
protected final List<String> logBuffer;
protected boolean logOutputIsScuccess = false;
/**
@ -134,9 +135,11 @@ public abstract class AbstractCommandExecutor {
processBuilder.redirectErrorStream(true);
// setting up user to run commands
command.add("sudo");
command.add("-u");
command.add(taskExecutionContext.getTenantCode());
if (CommonUtils.isSudoEnable()) {
command.add("sudo");
command.add("-u");
command.add(taskExecutionContext.getTenantCode());
}
command.add(commandInterpreter());
command.addAll(commandOptions());
command.add(commandFile);
@ -289,7 +292,7 @@ public abstract class AbstractCommandExecutor {
}
}
return process.isAlive();
return !process.isAlive();
}
/**
@ -595,4 +598,4 @@ public abstract class AbstractCommandExecutor {
public void setTaskResultString(String taskResultString) {
this.taskResultString = taskResultString;
}
}
}

6
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/MasterExecThreadTest.java
View File

@ -102,16 +102,12 @@ public class MasterExecThreadTest {
processDefinition.setGlobalParamList(Collections.EMPTY_LIST);
Mockito.when(processInstance.getProcessDefinition()).thenReturn(processDefinition);
masterExecThread = PowerMockito.spy(new MasterExecThread(
processInstance
, processService
, null, null, config));
masterExecThread = PowerMockito.spy(new MasterExecThread(processInstance, processService, null, null, config));
// prepareProcess init dag
Field dag = MasterExecThread.class.getDeclaredField("dag");
dag.setAccessible(true);
dag.set(masterExecThread, new DAG());
PowerMockito.doNothing().when(masterExecThread, "executeProcess");
PowerMockito.doNothing().when(masterExecThread, "postHandle");
PowerMockito.doNothing().when(masterExecThread, "prepareProcess");
PowerMockito.doNothing().when(masterExecThread, "runProcess");
PowerMockito.doNothing().when(masterExecThread, "endProcess");

39
dolphinscheduler-server/src/test/java/org/apache/dolphinscheduler/server/master/config/MasterConfigTest.java Normal file
View File

@ -0,0 +1,39 @@
/*
* 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.config;
import org.junit.Assert;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.test.context.junit4.SpringJUnit4ClassRunner;
@RunWith(SpringJUnit4ClassRunner.class)
@ContextConfiguration(classes = {MasterConfig.class})
public class MasterConfigTest {
@Autowired
private MasterConfig masterConfig;
@Test
public void getMasterDispatchTaskNumber() {
int masterDispatchTaskNumber = masterConfig.getMasterDispatchTaskNumber();
Assert.assertEquals(6, masterDispatchTaskNumber);
}
}

43
dolphinscheduler-server/src/test/resources/master.properties Normal file
View File

@ -0,0 +1,43 @@
#
# 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.
#
# master execute thread num
#master.exec.threads=100
# master execute task number in parallel
#master.exec.task.num=20
# master dispatch task number
master.dispatch.task.num=6
# master heartbeat interval
#master.heartbeat.interval=10
# master commit task retry times
#master.task.commit.retryTimes=5
# master commit task interval
#master.task.commit.interval=1000
# only less than cpu avg load, master server can work. default value -1 : the number of cpu cores * 2
#master.max.cpuload.avg=-1
# only larger than reserved memory, master server can work. default value : physical memory * 1/10, unit is G.
#master.reserved.memory=0.3
# master listen port
#master.listen.port=5678

2
dolphinscheduler-ui/src/js/conf/home/pages/projects/pages/instance/pages/gantt/index.vue
View File

@ -32,7 +32,7 @@
<template v-if="!isNodata">
<div class="gantt"></div>
</template>
<template v-if="isNodata">
<template v-else>
<m-no-data></m-no-data>
</template>
<m-spin :is-spin="isLoading">

3
dolphinscheduler-ui/src/js/conf/home/pages/resource/pages/file/pages/edit/index.vue
View File

@ -32,9 +32,8 @@
</div>
</template>
<m-no-data :msg="msg" v-if="msg"></m-no-data>
</template>
<template v-if="!isViewType">
<template v-else>
<m-no-type></m-no-type>
</template>
</div>

1
pom.xml
View File

@ -915,6 +915,7 @@
<!--<include>**/server/log/TaskLogDiscriminatorTest.java</include>-->
<include>**/server/log/TaskLogFilterTest.java</include>
<include>**/server/log/WorkerLogFilterTest.java</include>
<include>**/server/master/config/MasterConfigTest.java</include>
<include>**/server/master/consumer/TaskPriorityQueueConsumerTest.java</include>
<include>**/server/master/runner/MasterTaskExecThreadTest.java</include>
<!--<include>**/server/master/dispatch/executor/NettyExecutorManagerTest.java</include>-->