Skip to content
体验新版
项目
组织
正在加载...
登录
切换导航
打开侧边栏
xiphi1978
DolphinScheduler
提交
0cfa35da
DolphinScheduler
项目概览
xiphi1978
/
DolphinScheduler
与 Fork 源项目一致
Fork自
apache / DolphinScheduler
通知
2
Star
0
Fork
1
代码
文件
提交
分支
Tags
贡献者
分支图
Diff
Issue
0
列表
看板
标记
里程碑
合并请求
0
Wiki
0
Wiki
分析
仓库
DevOps
项目成员
Pages
DolphinScheduler
项目概览
项目概览
详情
发布
仓库
仓库
文件
提交
分支
标签
贡献者
分支图
比较
Issue
0
Issue
0
列表
看板
标记
里程碑
合并请求
0
合并请求
0
Pages
分析
分析
仓库分析
DevOps
Wiki
0
Wiki
成员
成员
收起侧边栏
关闭侧边栏
动态
分支图
创建新Issue
提交
Issue看板
体验新版 GitCode,发现更多精彩内容 >>
提交
0cfa35da
编写于
5月 21, 2019
作者:
G
gongzijian
浏览文件
操作
浏览文件
下载
差异文件
fix
上级
7b1f95d0
0cc61f5d
变更
10
隐藏空白更改
内联
并排
Showing
10 changed file
with
73 addition
and
33 deletion
+73
-33
escheduler-api/src/main/java/cn/escheduler/api/service/ResourcesService.java
...main/java/cn/escheduler/api/service/ResourcesService.java
+7
-5
escheduler-dao/src/main/java/cn/escheduler/dao/ProcessDao.java
...duler-dao/src/main/java/cn/escheduler/dao/ProcessDao.java
+48
-13
escheduler-dao/src/main/java/cn/escheduler/dao/model/TaskInstance.java
...o/src/main/java/cn/escheduler/dao/model/TaskInstance.java
+5
-1
escheduler-server/src/main/java/cn/escheduler/server/master/runner/MasterExecThread.java
.../cn/escheduler/server/master/runner/MasterExecThread.java
+1
-1
escheduler-server/src/main/java/cn/escheduler/server/utils/AlertManager.java
...rc/main/java/cn/escheduler/server/utils/AlertManager.java
+1
-1
escheduler-server/src/main/java/cn/escheduler/server/zk/ZKMasterClient.java
...src/main/java/cn/escheduler/server/zk/ZKMasterClient.java
+2
-2
escheduler-ui/.env
escheduler-ui/.env
+2
-3
escheduler-ui/install-escheduler-ui.sh
escheduler-ui/install-escheduler-ui.sh
+4
-4
escheduler-ui/src/js/module/i18n/locale/zh_CN.js
escheduler-ui/src/js/module/i18n/locale/zh_CN.js
+2
-2
script/env/.escheduler_env.sh
script/env/.escheduler_env.sh
+1
-1
未找到文件。
escheduler-api/src/main/java/cn/escheduler/api/service/ResourcesService.java
浏览文件 @
0cfa35da
...
...
@@ -701,17 +701,19 @@ public class ResourcesService extends BaseService {
if
(
checkAdmin
(
loginUser
,
result
))
{
return
result
;
}
List
<
Resource
>
resourceList
=
resourcesMapper
.
queryResourceExceptUserId
(
userId
);
Set
<
Resource
>
resourceSet
=
null
;
List
<
Object
>
list
;
if
(
resourceList
!=
null
&&
resourceList
.
size
()
>
0
)
{
resourceSet
=
new
HashSet
<>(
resourceList
);
Set
<
Resource
>
resourceSet
=
new
HashSet
<>(
resourceList
);
List
<
Resource
>
authedResourceList
=
resourcesMapper
.
queryAuthorizedResourceList
(
userId
);
getAuthorizedResourceList
(
resourceSet
,
authedResourceList
);
list
=
new
ArrayList
<>(
resourceSet
);
}
else
{
list
=
new
ArrayList
<>(
0
);
}
result
.
put
(
Constants
.
DATA_LIST
,
new
ArrayList
<>(
resourceSet
));
result
.
put
(
Constants
.
DATA_LIST
,
list
);
putMsg
(
result
,
Status
.
SUCCESS
);
return
result
;
}
...
...
escheduler-dao/src/main/java/cn/escheduler/dao/ProcessDao.java
浏览文件 @
0cfa35da
...
...
@@ -20,6 +20,7 @@ import cn.escheduler.common.Constants;
import
cn.escheduler.common.enums.*
;
import
cn.escheduler.common.model.DateInterval
;
import
cn.escheduler.common.model.TaskNode
;
import
cn.escheduler.common.process.Property
;
import
cn.escheduler.common.queue.ITaskQueue
;
import
cn.escheduler.common.queue.TaskQueueFactory
;
import
cn.escheduler.common.task.subprocess.SubProcessParameters
;
...
...
@@ -41,6 +42,7 @@ import org.springframework.stereotype.Component;
import
org.springframework.transaction.annotation.Transactional
;
import
java.util.*
;
import
java.util.stream.Collectors
;
import
static
cn
.
escheduler
.
common
.
Constants
.*;
import
static
cn
.
escheduler
.
dao
.
datasource
.
ConnectionFactory
.
getMapper
;
...
...
@@ -689,41 +691,62 @@ public class ProcessDao extends AbstractBaseDao {
* handle sub work process instance, update relation table and command parameters
* set sub work process flag, extends parent work process command parameters.
*/
public
ProcessInstance
setSubProcessParam
(
ProcessInstance
p
rocessInstance
){
String
cmdParam
=
p
rocessInstance
.
getCommandParam
();
public
ProcessInstance
setSubProcessParam
(
ProcessInstance
subP
rocessInstance
){
String
cmdParam
=
subP
rocessInstance
.
getCommandParam
();
if
(
StringUtils
.
isEmpty
(
cmdParam
)){
return
p
rocessInstance
;
return
subP
rocessInstance
;
}
Map
<
String
,
String
>
paramMap
=
JSONUtils
.
toMap
(
cmdParam
);
// write sub process id into cmd param.
if
(
paramMap
.
containsKey
(
CMDPARAM_SUB_PROCESS
)
&&
CMDPARAM_EMPTY_SUB_PROCESS
.
equals
(
paramMap
.
get
(
CMDPARAM_SUB_PROCESS
))){
paramMap
.
remove
(
CMDPARAM_SUB_PROCESS
);
paramMap
.
put
(
CMDPARAM_SUB_PROCESS
,
String
.
valueOf
(
p
rocessInstance
.
getId
()));
p
rocessInstance
.
setCommandParam
(
JSONUtils
.
toJson
(
paramMap
));
p
rocessInstance
.
setIsSubProcess
(
Flag
.
YES
);
this
.
saveProcessInstance
(
p
rocessInstance
);
paramMap
.
put
(
CMDPARAM_SUB_PROCESS
,
String
.
valueOf
(
subP
rocessInstance
.
getId
()));
subP
rocessInstance
.
setCommandParam
(
JSONUtils
.
toJson
(
paramMap
));
subP
rocessInstance
.
setIsSubProcess
(
Flag
.
YES
);
this
.
saveProcessInstance
(
subP
rocessInstance
);
}
// copy parent instance user def params to sub process..
String
parentInstanceId
=
paramMap
.
get
(
CMDPARAM_SUB_PROCESS_PARENT_INSTANCE_ID
);
if
(
StringUtils
.
isNotEmpty
(
parentInstanceId
)){
ProcessInstance
parentInstance
=
findProcessInstanceDetailById
(
Integer
.
parseInt
(
parentInstanceId
));
if
(
parentInstance
!=
null
){
processInstance
.
setGlobalParams
(
parentInstance
.
getGlobalParams
());
this
.
saveProcessInstance
(
processInstance
);
subProcessInstance
.
setGlobalParams
(
joinGlobalParams
(
parentInstance
.
getGlobalParams
(),
subProcessInstance
.
getGlobalParams
()));
this
.
saveProcessInstance
(
subProcessInstance
);
}
else
{
logger
.
error
(
"sub process command params error, cannot find parent instance: {} "
,
cmdParam
);
}
}
ProcessInstanceMap
processInstanceMap
=
JSONUtils
.
parseObject
(
cmdParam
,
ProcessInstanceMap
.
class
);
if
(
processInstanceMap
==
null
||
processInstanceMap
.
getParentProcessInstanceId
()
==
0
){
return
p
rocessInstance
;
return
subP
rocessInstance
;
}
// update sub process id to process map table
processInstanceMap
.
setProcessInstanceId
(
p
rocessInstance
.
getId
());
processInstanceMap
.
setProcessInstanceId
(
subP
rocessInstance
.
getId
());
this
.
updateWorkProcessInstanceMap
(
processInstanceMap
);
return
processInstance
;
return
subProcessInstance
;
}
/**
* join parent global params into sub process.
* only the keys doesn't in sub process global would be joined.
* @param parentGlobalParams
* @param subGlobalParams
* @return
*/
private
String
joinGlobalParams
(
String
parentGlobalParams
,
String
subGlobalParams
){
List
<
Property
>
parentPropertyList
=
JSONUtils
.
toList
(
parentGlobalParams
,
Property
.
class
);
List
<
Property
>
subPropertyList
=
JSONUtils
.
toList
(
subGlobalParams
,
Property
.
class
);
Map
<
String
,
String
>
subMap
=
subPropertyList
.
stream
().
collect
(
Collectors
.
toMap
(
Property:
:
getProp
,
Property:
:
getValue
));
for
(
Property
parent
:
parentPropertyList
){
if
(!
subMap
.
containsKey
(
parent
.
getProp
())){
subPropertyList
.
add
(
parent
);
}
}
return
JSONUtils
.
toJson
(
subPropertyList
);
}
/**
...
...
@@ -898,7 +921,11 @@ public class ProcessDao extends AbstractBaseDao {
taskInstance
.
setFlag
(
Flag
.
NO
);
updateTaskInstance
(
taskInstance
);
// crate new task instance
taskInstance
.
setRetryTimes
(
taskInstance
.
getRetryTimes
()
+
1
);
if
(
taskInstance
.
getState
()
!=
ExecutionStatus
.
NEED_FAULT_TOLERANCE
){
taskInstance
.
setRetryTimes
(
taskInstance
.
getRetryTimes
()
+
1
);
}
taskInstance
.
setEndTime
(
null
);
taskInstance
.
setStartTime
(
new
Date
());
taskInstance
.
setFlag
(
Flag
.
YES
);
taskInstance
.
setHost
(
null
);
taskInstance
.
setId
(
0
);
...
...
@@ -1526,6 +1553,14 @@ public class ProcessDao extends AbstractBaseDao {
}
public
void
selfFaultTolerant
(
int
...
states
){
List
<
ProcessInstance
>
processInstanceList
=
processInstanceMapper
.
listByStatus
(
states
);
for
(
ProcessInstance
processInstance:
processInstanceList
){
selfFaultTolerant
(
processInstance
);
}
}
@Transactional
(
value
=
"TransactionManager"
,
rollbackFor
=
Exception
.
class
)
public
void
selfFaultTolerant
(
ProcessInstance
processInstance
){
...
...
escheduler-dao/src/main/java/cn/escheduler/dao/model/TaskInstance.java
浏览文件 @
0cfa35da
...
...
@@ -422,8 +422,12 @@ public class TaskInstance {
if
(
this
.
isSubProcess
()){
return
false
;
}
return
(
this
.
getState
().
typeIsFailure
()
if
(
this
.
getState
()
==
ExecutionStatus
.
NEED_FAULT_TOLERANCE
){
return
true
;
}
else
{
return
(
this
.
getState
().
typeIsFailure
()
&&
this
.
getRetryTimes
()
<
this
.
getMaxRetryTimes
());
}
}
public
void
setDependency
(
String
dependency
)
{
...
...
escheduler-server/src/main/java/cn/escheduler/server/master/runner/MasterExecThread.java
浏览文件 @
0cfa35da
...
...
@@ -869,7 +869,7 @@ public class MasterExecThread implements Runnable {
}
Date
now
=
new
Date
();
long
runningTime
=
DateUtils
.
diff
erMs
(
now
,
processInstance
.
getStartTime
());
long
runningTime
=
DateUtils
.
diff
Min
(
now
,
processInstance
.
getStartTime
());
if
(
runningTime
>
processInstance
.
getTimeout
()){
return
true
;
...
...
escheduler-server/src/main/java/cn/escheduler/server/utils/AlertManager.java
浏览文件 @
0cfa35da
...
...
@@ -175,7 +175,7 @@ public class AlertManager {
alert
.
setContent
(
content
);
alert
.
setAlertType
(
AlertType
.
EMAIL
);
alert
.
setCreateTime
(
new
Date
());
alert
.
setAlertGroupId
(
processInstance
.
getWarningGroupId
());
alert
.
setAlertGroupId
(
processInstance
.
getWarningGroupId
()
==
null
?
1
:
processInstance
.
getWarningGroupId
()
);
alert
.
setReceivers
(
processInstance
.
getProcessDefinition
().
getReceivers
());
alert
.
setReceiversCc
(
processInstance
.
getProcessDefinition
().
getReceiversCc
());
...
...
escheduler-server/src/main/java/cn/escheduler/server/zk/ZKMasterClient.java
浏览文件 @
0cfa35da
...
...
@@ -123,9 +123,9 @@ public class ZKMasterClient extends AbstractZKClient {
// register master
this
.
registMaster
();
// check if fault tolerance is required
// check if fault tolerance is required
,failure and tolerance
if
(
getActiveMasterNum
()
==
1
)
{
processDao
.
selfFaultTolerant
(
ExecutionStatus
.
RUNNING_EXEUTION
.
ordinal
());
processDao
.
selfFaultTolerant
(
ExecutionStatus
.
RUNNING_EXEUTION
.
ordinal
()
,
ExecutionStatus
.
NEED_FAULT_TOLERANCE
.
ordinal
()
);
}
}
...
...
escheduler-ui/.env
浏览文件 @
0cfa35da
# 后端接口地址
#API_BASE = http://192.168.220.154:12345
API_BASE = http://192.168.221.188:12345
# 后端接口地址11
API_BASE = http://192.168.xx.xx:12345
# 本地开发如需ip访问项目把"#"号去掉
#DEV_HOST = 192.168.xx.xx
escheduler-ui/install-escheduler-ui.sh
浏览文件 @
0cfa35da
...
...
@@ -88,9 +88,9 @@ eschedulerConf(){
proxy_set_header remote_addr
$E_remote_addr
;
proxy_set_header X-Forwarded-For
$E_proxy_add_x_forwarded_for
;
proxy_http_version 1.1;
proxy_connect_timeout
4
s;
proxy_read_timeout 30s;
proxy_send_timeout
12
s;
proxy_connect_timeout
300
s;
proxy_read_timeout 30
0
s;
proxy_send_timeout
300
s;
proxy_set_header Upgrade
$E_http_upgrade
;
proxy_set_header Connection "
upgrade
";
}
...
...
@@ -184,7 +184,7 @@ if [[ $version -eq 6 ]]; then
fi
# centos 7
if
[[
$version
-eq
7
]]
;
then
if
[[
$version
-eq
7
]]
;
then
centos7
fi
...
...
escheduler-ui/src/js/module/i18n/locale/zh_CN.js
浏览文件 @
0cfa35da
...
...
@@ -130,7 +130,7 @@ export default {
'
Please enter database name
'
:
'
请输入数据库名
'
,
'
jdbc connect parameters
'
:
'
jdbc连接参数
'
,
'
Test Connect
'
:
'
测试连接
'
,
'
Please enter resource name
'
:
'
请输入
资
源名称
'
,
'
Please enter resource name
'
:
'
请输入
数据
源名称
'
,
'
Please enter IP/hostname
'
:
'
请输入IP/主机名
'
,
'
jdbc connection parameters is not a correct JSON format
'
:
'
jdbc连接参数不是一个正确的JSON格式
'
,
'
#
'
:
'
编号
'
,
...
...
@@ -419,7 +419,7 @@ export default {
'
Create token
'
:
'
创建令牌
'
,
'
Edit token
'
:
'
编辑令牌
'
,
'
Please enter the IP address separated by commas
'
:
'
请输入IP地址多个用逗号隔开
'
,
'
Note: Multiple IP addresses have been comma separated
'
:
'
注意:多个IP地址
已
逗号分割
'
,
'
Note: Multiple IP addresses have been comma separated
'
:
'
注意:多个IP地址
以
逗号分割
'
,
'
Failure time
'
:
'
失效时间
'
,
'
User
'
:
'
用户
'
,
'
Please enter token
'
:
'
请输入令牌
'
,
...
...
script/env/.escheduler_env.sh
浏览文件 @
0cfa35da
...
...
@@ -6,4 +6,4 @@ export PYTHON_HOME=/opt/soft/python
export
JAVA_HOME
=
/opt/soft/java
export
HIVE_HOME
=
/opt/soft/hive
export
PATH
=
$HADOOP_HOME
/bin:
$SPARK_HOME1
/bin:
$SPARK_HOME2
/bin:
$PYTHON_HOME
/bin:
$JAVA_HOME
/bin:
$HIVE_HOME
/bin:
$PATH
\ No newline at end of file
export
PATH
=
$HADOOP_HOME
/bin:
$SPARK_HOME1
/bin:
$SPARK_HOME2
/bin:
$PYTHON_HOME
:
$JAVA_HOME
/bin:
$HIVE_HOME
/bin:
$PATH
\ No newline at end of file
编辑
预览
Markdown
is supported
0%
请重试
或
添加新附件
.
添加附件
取消
You are about to add
0
people
to the discussion. Proceed with caution.
先完成此消息的编辑!
取消
想要评论请
注册
或
登录