Compare commits

...

21 Commits

Author SHA1 Message Date
Jay Chung bcf6b18e0f
merge schema 330 into 321 and change docs (#15582) 2024-02-06 23:48:58 +08:00
seckiller 58aba58079 [fix][worker]:Fix the issue of missing disk usage reporting information in the worker (#15565)
Co-authored-by: Rick Cheng <rickchengx@gmail.com>
(cherry picked from commit 89da67ddeb)
2024-02-06 18:25:01 +08:00
Wenjun Ruan 9bb526855d Set the workflow instance ready state to running in failover (#15572)
(cherry picked from commit 1b32e28cc8)
2024-02-06 18:24:58 +08:00
旺阳 0b722999d3 [Fix] Fix WorkflowInstance batch start failed will throw incorrect exception. (#15577)
(cherry picked from commit 73a5a77db2)
2024-02-06 18:24:50 +08:00
Wenjun Ruan eff7364ee1 Fix create parent directory will cause FileAlreadyExistsException (#15576)
Co-authored-by: Rick Cheng <rickchengx@gmail.com>
(cherry picked from commit 4541e74e53)
2024-02-06 18:24:47 +08:00
John Huang 8e9537ad33 [Improvement][UT] Improve Worker registry coverage (#15380)
Co-authored-by: fuchanghai <changhaifu@apache.org>
Co-authored-by: Eric Gao <ericgao.apache@gmail.com>
Co-authored-by: Rick Cheng <rickchengx@gmail.com>
(cherry picked from commit 69676b445c)
2024-02-06 18:24:42 +08:00
Wenjun Ruan 52d315d3ca Fix Recover WorkflowInstance will casue workflow Instance state is success but task insatnce is killed/paused (#15574)
(cherry picked from commit 43a06525a2)
2024-02-06 18:24:39 +08:00
Jay Chung e2d08fc27d fix: data quality may fail in docker mode (#15563)
(cherry picked from commit 91d56f4860)
2024-02-06 18:24:33 +08:00
Jay Chung fa7b4c1c8f fix: start param for wf not work (#15544)
* fix: start param for wf not work

fix: #15280

* fix test

(cherry picked from commit 01eb8f834f)
2024-02-06 18:24:29 +08:00
小可耐 7d660ec2f9 [Bug][force-success] force success add end time (#15144)
Co-authored-by: Rick Cheng <rickchengx@gmail.com>
(cherry picked from commit e5a208f363)
2024-02-06 18:24:26 +08:00
Jay Chung d87a1acd19 fix: ddl without drop exists (#14128)
Co-authored-by: Rick Cheng <rickchengx@gmail.com>
(cherry picked from commit d8e820c4b9)
2024-02-06 18:23:03 +08:00
dada Sun 19adf25920 [Bug][Task Api] fix 'MACPATTERN' in ProcessUtils and cover all cases on MacOS in ProcessUtilsTest (#15480) (#15529)
Co-authored-by: Rick Cheng <rickchengx@gmail.com>
(cherry picked from commit 4d6af516b9)
2024-02-06 18:22:58 +08:00
caishunfeng ca93aa8e26 fix switch js (#15487)
Co-authored-by: Rick Cheng <rickchengx@gmail.com>
Co-authored-by: Eric Gao <ericgao.apache@gmail.com>
(cherry picked from commit ef9ed3db55)
2024-02-06 18:22:55 +08:00
Jay Chung acb1e0a09f fix: data quality can not use (#15551)
* fix: data quality can not use

fix: #15468, #15249, #14858

---------

Co-authored-by: Rick Cheng <rickchengx@gmail.com>
Co-authored-by: Eric Gao <ericgao.apache@gmail.com>
(cherry picked from commit 8efaa9fa1f)
2024-02-06 18:22:51 +08:00
Wenjun Ruan a76b4662be Fix createFile with permission will not work (#15556)
(cherry picked from commit 4d15932ede)
2024-02-06 18:22:47 +08:00
John Huang 9e4e890f84 [Doc][Docker] fix typo on start with docker (#15534)
Co-authored-by: xiangzihao <460888207@qq.com>
Co-authored-by: Eric Gao <ericgao.apache@gmail.com>
(cherry picked from commit 1b42d45fcd)
2024-02-06 18:22:42 +08:00
Wenjun Ruan c8f120a60a Fix TaskGroupQueue will never be wakeup due to wakeup failed at one time (#15528)
(cherry picked from commit 89742332a8)
2024-02-06 18:22:34 +08:00
John Huang d9529d95f3 refactor comments & function name for confuse (#15546)
Co-authored-by: fuchanghai <changhaifu@apache.org>
(cherry picked from commit 2f66a667ba)
2024-02-06 18:22:32 +08:00
Wenjun Ruan be58ba69a9 Exit JVM when OOM (#15538)
(cherry picked from commit 25810a81dc)
2024-02-06 18:21:56 +08:00
Jay Chung 898b001608
cp: Reduce the size of tarball to continue ASF release (#15004) (#15540)
* Reduce the size of tarball to continue ASF release

for more detail you can see https://lists.apache.org/thread/rmp7fghlj0n7h9y2v3p8gkw9f9qbo6qt
2024-02-06 18:15:57 +08:00
Jay Chung 67615bd815
chore: Docs change for 3.2.1 release (#15539) 2024-01-31 16:45:54 +08:00
152 changed files with 3317 additions and 2150 deletions

View File

@ -149,7 +149,7 @@ jobs:
fail-fast: false
matrix:
db: ["mysql", "postgresql"]
version: ["2.0.9", "3.0.6", "3.1.8", "3.2.0"]
version: ["2.0.9", "3.0.6", "3.1.9", "3.2.0"]
steps:
- name: Set up JDK 8
uses: actions/setup-java@v2

View File

@ -15,8 +15,8 @@
# specific language governing permissions and limitations
# under the License.
#
HUB=ghcr.io/apache/dolphinscheduler
TAG=latest
HUB=apache
TAG=3.2.1
TZ=Asia/Shanghai
DATABASE=postgresql

View File

@ -35,11 +35,11 @@ type: application
# This is the chart version. This version number should be incremented each time you make changes
# to the chart and its templates, including the app version.
version: 3.1.0
version: 3.2.1
# This is the version number of the application being deployed. This version number should be
# incremented each time you make changes to the application.
appVersion: 3.1.0
appVersion: 3.2.1
dependencies:
- name: postgresql

View File

@ -121,7 +121,7 @@ Please refer to the [Quick Start in Kubernetes](../../../docs/docs/en/guide/inst
| conf.common."alert.rpc.port" | int | `50052` | rpc port |
| conf.common."appId.collect" | string | `"log"` | way to collect applicationId: log, aop |
| conf.common."conda.path" | string | `"/opt/anaconda3/etc/profile.d/conda.sh"` | set path of conda.sh |
| conf.common."data-quality.jar.name" | string | `"dolphinscheduler-data-quality-dev-SNAPSHOT.jar"` | data quality option |
| conf.common."data-quality.jar.dir" | string | `nil` | data quality option |
| conf.common."data.basedir.path" | string | `"/tmp/dolphinscheduler"` | user data local directory path, please make sure the directory exists and have read write permissions |
| conf.common."datasource.encryption.enable" | bool | `false` | datasource encryption enable |
| conf.common."datasource.encryption.salt" | string | `"!@#$%^&*"` | datasource encryption salt |
@ -175,7 +175,7 @@ Please refer to the [Quick Start in Kubernetes](../../../docs/docs/en/guide/inst
| image.pullPolicy | string | `"IfNotPresent"` | Image pull policy. Options: Always, Never, IfNotPresent |
| image.pullSecret | string | `""` | Specify a imagePullSecrets |
| image.registry | string | `"apache/dolphinscheduler"` | Docker image repository for the DolphinScheduler |
| image.tag | string | `"latest"` | Docker image version for the DolphinScheduler |
| image.tag | string | `"3.2.1"` | Docker image version for the DolphinScheduler |
| image.tools | string | `"dolphinscheduler-tools"` | tools image |
| image.worker | string | `"dolphinscheduler-worker"` | worker image |
| ingress.annotations | object | `{}` | Ingress annotations |

View File

@ -33,7 +33,7 @@ image:
# -- Docker image repository for the DolphinScheduler
registry: apache/dolphinscheduler
# -- Docker image version for the DolphinScheduler
tag: latest
tag: 3.2.1
# -- Image pull policy. Options: Always, Never, IfNotPresent
pullPolicy: "IfNotPresent"
# -- Specify a imagePullSecrets
@ -328,7 +328,7 @@ conf:
datasource.encryption.salt: '!@#$%^&*'
# -- data quality option
data-quality.jar.name: dolphinscheduler-data-quality-dev-SNAPSHOT.jar
data-quality.jar.dir:
# -- Whether hive SQL is executed in the same session
support.hive.oneSession: false

File diff suppressed because it is too large Load Diff

View File

@ -226,7 +226,7 @@ The default configuration is as follows:
| yarn.job.history.status.address | http://ds1:19888/ws/v1/history/mapreduce/jobs/%s | job history status url of yarn |
| datasource.encryption.enable | false | whether to enable datasource encryption |
| datasource.encryption.salt | !@#$%^&* | the salt of the datasource encryption |
| data-quality.jar.name | dolphinscheduler-data-quality-dev-SNAPSHOT.jar | the jar of data quality |
| data-quality.jar.dir | | the jar of data quality |
| support.hive.oneSession | false | specify whether hive SQL is executed in the same session |
| sudo.enable | true | whether to enable sudo |
| alert.rpc.port | 50052 | the RPC port of Alert Server |

View File

@ -12,15 +12,7 @@ The execution logic of the data quality task is as follows:
- The current data quality task result is stored in the `t_ds_dq_execute_result` table of `dolphinscheduler`
`Worker` sends the task result to `Master`, after `Master` receives `TaskResponse`, it will judge whether the task type is `DataQualityTask`, if so, it will read the corresponding result from `t_ds_dq_execute_result` according to `taskInstanceId`, and then The result is judged according to the check mode, operator and threshold configured by the user.
- If the result is a failure, the corresponding operation, alarm or interruption will be performed according to the failure policy configured by the user.
- Add config : `<server-name>/conf/common.properties`
```properties
# Change to specific version if you not use dev branch
data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
```
- Please fill in `data-quality.jar.name` according to the actual package name.
- If you package `data-quality` separately, remember to modify the package name to be consistent with `data-quality.jar.name`.
- If you package `data-quality` separately, remember to modify the package name to be consistent with `data-quality.jar.dir` in `common.properties` with attribute name `data-quality.jar.dir`
- If the old version is upgraded and used, you need to execute the `sql` update script to initialize the database before running.
- `dolphinscheduler-data-quality-dev-SNAPSHOT.jar` was built with no dependencies. If a `JDBC` driver is required, you can set the `-jars` parameter in the `node settings` `Option Parameters`, e.g. `--jars /lib/jars/mysql-connector-java-8.0.16.jar`.
- Currently only `MySQL`, `PostgreSQL` and `HIVE` data sources have been tested, other data sources have not been tested yet.

View File

@ -30,9 +30,9 @@ Attention: DolphinScheduler itself does not depend on Hadoop, Hive, Spark, but w
mkdir -p /opt
cd /opt
# decompress
tar -zxvf apache-dolphinscheduler-<version>-bin.tar.gz -C /opt
tar -zxvf apache-dolphinscheduler-3.2.1-bin.tar.gz -C /opt
cd /opt
mv apache-dolphinscheduler-<version>-bin dolphinscheduler
mv apache-dolphinscheduler-3.2.1-bin dolphinscheduler
```
```markdown

View File

@ -14,16 +14,16 @@ If you are a new hand and want to experience DolphinScheduler functions, we reco
## Install DolphinScheduler
Please download the source code package `apache-dolphinscheduler-<version>-src.tar.gz`, download address: [download address](https://dolphinscheduler.apache.org/en-us/download)
Please download the source code package `apache-dolphinscheduler-3.2.1-src.tar.gz`, download address: [download address](https://dolphinscheduler.apache.org/en-us/download)
To publish the release name `dolphinscheduler` version, please execute the following commands:
```
$ tar -zxvf apache-dolphinscheduler-<version>-src.tar.gz
$ cd apache-dolphinscheduler-<version>-src/deploy/kubernetes/dolphinscheduler
$ tar -zxvf apache-dolphinscheduler-3.2.1-src.tar.gz
$ cd apache-dolphinscheduler-3.2.1-src/deploy/kubernetes/dolphinscheduler
$ helm repo add bitnami https://charts.bitnami.com/bitnami
$ helm dependency update .
$ helm install dolphinscheduler . --set image.tag=<version>
$ helm install dolphinscheduler . --set image.tag=3.2.1
```
To publish the release name `dolphinscheduler` version to `test` namespace:
@ -234,9 +234,9 @@ kubectl scale --replicas=6 sts dolphinscheduler-worker -n test # with test names
2. Create a new `Dockerfile` to add MySQL driver:
```
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-<service>:<version>
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-<service>:3.2.1
# For example
# FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-tools:<version>
# FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-tools:3.2.1
# Attention Please, If the build is dolphinscheduler-tools image
# You need to change the following line to: COPY mysql-connector-java-8.0.16.jar /opt/dolphinscheduler/tools/libs
@ -285,9 +285,9 @@ externalDatabase:
2. Create a new `Dockerfile` to add MySQL or Oracle driver:
```
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-<service>:<version>
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-<service>:3.2.1
# For example
# FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:<version>
# FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:3.2.1
# If you want to support MySQL Datasource
COPY mysql-connector-java-8.0.16.jar /opt/dolphinscheduler/libs
@ -317,7 +317,7 @@ docker build -t apache/dolphinscheduler-<service>:new-driver .
1. Create a new `Dockerfile` to install pip:
```
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:<version>
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:3.2.1
COPY requirements.txt /tmp
RUN apt-get update && \
apt-get install -y --no-install-recommends python-pip && \
@ -352,7 +352,7 @@ docker build -t apache/dolphinscheduler-worker:pip .
1. Create a new `Dockerfile` to install Python 3:
```
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:<version>
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:3.2.1
RUN apt-get update && \
apt-get install -y --no-install-recommends python3 && \
rm -rf /var/lib/apt/lists/*

View File

@ -12,7 +12,7 @@ If you want to deploy DolphinScheduler in production, we recommend you follow [c
## Preparation
- JDKdownload [JDK][jdk] (1.8+), install and configure environment variable `JAVA_HOME` and append `bin` dir (included in `JAVA_HOME`) to `PATH` variable. You can skip this step if it already exists in your environment.
- Binary package: download the DolphinScheduler binary package at [download page](https://dolphinscheduler.apache.org/en-us/download/<version>). <!-- markdown-link-check-disable-line -->
- Binary package: download the DolphinScheduler binary package at [download page](https://dolphinscheduler.apache.org/en-us/download/3.2.1). <!-- markdown-link-check-disable-line -->
## Start DolphinScheduler Standalone Server

View File

@ -152,8 +152,10 @@ datasource.encryption.enable=false
# datasource encryption salt
datasource.encryption.salt=!@#$%^&*
# data quality option
data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
# data quality jar directory path, it would auto discovery data quality jar from this given dir. You should keep it empty if you do not change anything in
# data-quality, it will auto discovery by dolphinscheduler itself. Change it only if you want to use your own data-quality jar and it is not in worker-server
# libs directory(but may sure your jar name start with `dolphinscheduler-data-quality`).
data-quality.jar.dir=
#data-quality.error.output.path=/tmp/data-quality-error-data

View File

@ -19,7 +19,7 @@ Start DolphinScheduler with standalone-server Docker images is the easiest way t
you can learn DolphinScheduler's concepts and usage, with minimal cost.
```shell
$ DOLPHINSCHEDULER_VERSION=<version>
$ DOLPHINSCHEDULER_VERSION=3.2.1
$ docker run --name dolphinscheduler-standalone-server -p 12345:12345 -p 25333:25333 -d apache/dolphinscheduler-standalone-server:"${DOLPHINSCHEDULER_VERSION}"
```
@ -37,11 +37,11 @@ be stored on disks after you change docker-compose configuration, and it is robu
DolphinScheduler in a long term. You have to install [docker-compose](https://docs.docker.com/compose/install/) before you
start servers.
After complete the installation, get the `docker-compose.yaml` file from [download page](https://dolphinscheduler.apache.org/en-us/download/<version>)
After complete the installation, get the `docker-compose.yaml` file from [download page](https://dolphinscheduler.apache.org/en-us/download/3.2.1)
form its source package, and make sure you get the right version. After download the package, you can run the commands as below.
```shell
$ DOLPHINSCHEDULER_VERSION=<version>
$ DOLPHINSCHEDULER_VERSION=3.2.1
$ tar -zxf apache-dolphinscheduler-"${DOLPHINSCHEDULER_VERSION}"-src.tar.gz
# Going to docker-compose's location
# For Mac or Linux users
@ -68,7 +68,7 @@ $ docker-compose --profile all up -d
container when it up. You could start DolphinScheduler server separately if you want to reuse your exists services.
```shell
$ DOLPHINSCHEDULER_VERSION=<version>
$ DOLPHINSCHEDULER_VERSION=3.2.1
# Initialize the database, make sure database <DATABASE> already exists
$ docker run -d --name dolphinscheduler-tools \
-e DATABASE="postgresql" \
@ -81,7 +81,7 @@ $ docker run -d --name dolphinscheduler-tools \
# Starting DolphinScheduler service
$ docker run -d --name dolphinscheduler-master \
-e DATABASE="postgresql" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/dolphinscheduler" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/<DATABASE>" \
-e SPRING_DATASOURCE_USERNAME="<USER>" \
-e SPRING_DATASOURCE_PASSWORD="<PASSWORD>" \
-e SPRING_JACKSON_TIME_ZONE="UTC" \
@ -90,7 +90,7 @@ $ docker run -d --name dolphinscheduler-master \
-d apache/dolphinscheduler-master:"${DOLPHINSCHEDULER_VERSION}"
$ docker run -d --name dolphinscheduler-worker \
-e DATABASE="postgresql" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/dolphinscheduler" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/<DATABASE>" \
-e SPRING_DATASOURCE_USERNAME="<USER>" \
-e SPRING_DATASOURCE_PASSWORD="<PASSWORD>" \
-e SPRING_JACKSON_TIME_ZONE="UTC" \
@ -99,7 +99,7 @@ $ docker run -d --name dolphinscheduler-worker \
-d apache/dolphinscheduler-worker:"${DOLPHINSCHEDULER_VERSION}"
$ docker run -d --name dolphinscheduler-api \
-e DATABASE="postgresql" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/dolphinscheduler" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/<DATABASE>" \
-e SPRING_DATASOURCE_USERNAME="<USER>" \
-e SPRING_DATASOURCE_PASSWORD="<PASSWORD>" \
-e SPRING_JACKSON_TIME_ZONE="UTC" \
@ -108,7 +108,7 @@ $ docker run -d --name dolphinscheduler-api \
-d apache/dolphinscheduler-api:"${DOLPHINSCHEDULER_VERSION}"
$ docker run -d --name dolphinscheduler-alert-server \
-e DATABASE="postgresql" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/dolphinscheduler" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/<DATABASE>" \
-e SPRING_DATASOURCE_USERNAME="<USER>" \
-e SPRING_DATASOURCE_PASSWORD="<PASSWORD>" \
-e SPRING_JACKSON_TIME_ZONE="UTC" \
@ -134,5 +134,5 @@ and use `admin` and `dolphinscheduler123` as default username and password in th
You can modify some environment variables to change configurations when you are starting servers through Docker. We have
an example in [using exists PostgreSQL ZooKeeper](#using-exists-postgresql-zookeeper) to change database and ZooKeeper configurations,
and you could find all environment variables in [all environment variables](https://github.com/apache/dolphinscheduler/blob/<version>/script/env/dolphinscheduler_env.sh) <!-- markdown-link-check-disable-line -->
and you could find all environment variables in [all environment variables](https://github.com/apache/dolphinscheduler/blob/3.2.1/script/env/dolphinscheduler_env.sh) <!-- markdown-link-check-disable-line -->
and change them if you want.

View File

@ -10,6 +10,8 @@ This document records the incompatible updates between each version. You need to
* Remove the spark version of spark task ([#11860](https://github.com/apache/dolphinscheduler/pull/11860)).
* Change the default unix shell executor from sh to bash ([#12180](https://github.com/apache/dolphinscheduler/pull/12180)).
* Remove `deleteSource` in `download()` of `StorageOperate` ([#14084](https://github.com/apache/dolphinscheduler/pull/14084))
* Remove default key for attribute `data-quality.jar.name` in `common.properties` ([#15551](https://github.com/apache/dolphinscheduler/pull/15551))
* Rename attribute `data-quality.jar.name` to `data-quality.jar.dir` in `common.properties` and represent for directory ([#15563](https://github.com/apache/dolphinscheduler/pull/15563))
## 3.2.0

View File

@ -4,6 +4,10 @@
#### Setup instructions, are available for each stable version of Apache DolphinScheduler below:
### Versions: 3.2.1
#### Links [3.2.1 Document](../3.2.1/user_doc/about/introduction.md)
### Versions: 3.2.0
#### Links [3.2.0 Document](../3.2.0/user_doc/about/introduction.md)

View File

@ -226,7 +226,7 @@ common.properties配置文件目前主要是配置hadoop/s3/yarn/applicationId
| yarn.job.history.status.address | http://ds1:19888/ws/v1/history/mapreduce/jobs/%s | yarn的作业历史状态URL |
| datasource.encryption.enable | false | 是否启用datasource 加密 |
| datasource.encryption.salt | !@#$%^&* | datasource加密使用的salt |
| data-quality.jar.name | dolphinscheduler-data-quality-dev-SNAPSHOT.jar | 配置数据质量使用的jar包 |
| data-quality.jar.dir | | 配置数据质量使用的jar包 |
| support.hive.oneSession | false | 设置hive SQL是否在同一个session中执行 |
| sudo.enable | true | 是否开启sudo |
| alert.rpc.port | 50052 | Alert Server的RPC端口 |

View File

@ -13,14 +13,7 @@
>
## 注意事项
添加配置信息:`<server-name>/conf/common.properties`
```properties
data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
```
- 这里的`data-quality.jar.name`请根据实际打包的名称来填写。
- 如果单独打包`data-quality`的话,记得修改包名和`data-quality.jar.name`一致。
- 如果单独打包`data-quality`的话,记得修改包路径和`data-quality.jar.dir`一致,配置内容在 `common.properties` 中的 `data-quality.jar.dir`
- 如果是老版本升级使用,运行之前需要先执行`SQL`更新脚本进行数据库初始化。
- 当前 `dolphinscheduler-data-quality-dev-SNAPSHOT.jar` 是瘦包,不包含任何 `JDBC` 驱动。
如果有 `JDBC` 驱动需要,可以在`节点设置` `选项参数`处设置 `--jars` 参数,

View File

@ -30,9 +30,9 @@
mkdir -p /opt
cd /opt
# 解压缩
tar -zxvf apache-dolphinscheduler-<version>-bin.tar.gz -C /opt
tar -zxvf apache-dolphinscheduler-3.2.1-bin.tar.gz -C /opt
cd /opt
mv apache-dolphinscheduler-<version>-bin dolphinscheduler
mv apache-dolphinscheduler-3.2.1-bin dolphinscheduler
```
```markdown

View File

@ -14,16 +14,16 @@ Kubernetes 部署目的是在 Kubernetes 集群中部署 DolphinScheduler 服务
## 安装 dolphinscheduler
请下载源码包 apache-dolphinscheduler-<version>-src.tar.gz下载地址: [下载](https://dolphinscheduler.apache.org/zh-cn/download)
请下载源码包 apache-dolphinscheduler-3.2.1-src.tar.gz下载地址: [下载](https://dolphinscheduler.apache.org/zh-cn/download)
发布一个名为 `dolphinscheduler` 的版本(release),请执行以下命令:
```
$ tar -zxvf apache-dolphinscheduler-<version>-src.tar.gz
$ cd apache-dolphinscheduler-<version>-src/deploy/kubernetes/dolphinscheduler
$ tar -zxvf apache-dolphinscheduler-3.2.1-src.tar.gz
$ cd apache-dolphinscheduler-3.2.1-src/deploy/kubernetes/dolphinscheduler
$ helm repo add bitnami https://charts.bitnami.com/bitnami
$ helm dependency update .
$ helm install dolphinscheduler . --set image.tag=<version>
$ helm install dolphinscheduler . --set image.tag=3.2.1
```
将名为 `dolphinscheduler` 的版本(release) 发布到 `test` 的命名空间中:
@ -233,9 +233,9 @@ kubectl scale --replicas=6 sts dolphinscheduler-worker -n test # with test names
2. 创建一个新的 `Dockerfile`,用于添加 MySQL 的驱动包:
```
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-<service>:<version>
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-<service>:3.2.1
# 例如
# FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-tools:<version>
# FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-tools:3.2.1
# 注意如果构建的是dolphinscheduler-tools镜像
# 需要将下面一行修改为COPY mysql-connector-java-8.0.16.jar /opt/dolphinscheduler/tools/libs
@ -284,9 +284,9 @@ externalDatabase:
2. 创建一个新的 `Dockerfile`,用于添加 MySQL 或者 Oracle 驱动包:
```
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-<service>:<version>
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-<service>:3.2.1
# 例如
# FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:<version>
# FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:3.2.1
# 如果你想支持 MySQL 数据源
COPY mysql-connector-java-8.0.16.jar /opt/dolphinscheduler/libs
@ -316,7 +316,7 @@ docker build -t apache/dolphinscheduler-<service>:new-driver .
1. 创建一个新的 `Dockerfile`,用于安装 pip:
```
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:<version>
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:3.2.1
COPY requirements.txt /tmp
RUN apt-get update && \
apt-get install -y --no-install-recommends python-pip && \
@ -351,7 +351,7 @@ docker build -t apache/dolphinscheduler-worker:pip .
1. 创建一个新的 `Dockerfile`,用于安装 Python 3:
```
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:<version>
FROM dolphinscheduler.docker.scarf.sh/apache/dolphinscheduler-worker:3.2.1
RUN apt-get update && \
apt-get install -y --no-install-recommends python3 && \
rm -rf /var/lib/apt/lists/*

View File

@ -10,7 +10,7 @@ Standalone 仅适用于 DolphinScheduler 的快速体验.
## 前置准备工作
- JDK下载[JDK][jdk] (1.8+),安装并配置 `JAVA_HOME` 环境变量,并将其下的 `bin` 目录追加到 `PATH` 环境变量中。如果你的环境中已存在,可以跳过这步。
- 二进制包:在[下载页面](https://dolphinscheduler.apache.org/en-us/download/<version>)下载 DolphinScheduler 二进制包 <!-- markdown-link-check-disable-line -->
- 二进制包:在[下载页面](https://dolphinscheduler.apache.org/en-us/download/3.2.1)下载 DolphinScheduler 二进制包 <!-- markdown-link-check-disable-line -->
## 启动 DolphinScheduler Standalone Server

View File

@ -156,8 +156,10 @@ datasource.encryption.enable=false
# datasource encryption salt
datasource.encryption.salt=!@#$%^&*
# data quality option
data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
# data quality jar directory path, it would auto discovery data quality jar from this given dir. You should keep it empty if you do not change anything in
# data-quality, it will auto discovery by dolphinscheduler itself. Change it only if you want to use your own data-quality jar and it is not in worker-server
# libs directory(but may sure your jar name start with `dolphinscheduler-data-quality`).
data-quality.jar.dir=
#data-quality.error.output.path=/tmp/data-quality-error-data

View File

@ -18,7 +18,7 @@
你可以最快速的体验到 DolphinScheduler 的大部分功能,了解主要和概念和内容。
```shell
$ DOLPHINSCHEDULER_VERSION=<version>
$ DOLPHINSCHEDULER_VERSION=3.2.1
$ docker run --name dolphinscheduler-standalone-server -p 12345:12345 -p 25333:25333 -d apache/dolphinscheduler-standalone-server:"${DOLPHINSCHEDULER_VERSION}"
```
@ -33,11 +33,11 @@ $ docker run --name dolphinscheduler-standalone-server -p 12345:12345 -p 25333:2
服务重启的时候保留元数据(如需要挂载到本地路径需要做指定)。他更健壮,能保证用户体验更加完整的 DolphinScheduler 服务。这种方式需要先安装
[docker-compose](https://docs.docker.com/compose/install/),链接适用于 MacLinuxWindows。
确保 docker-compose 顺利安装后,需要获取 `docker-compose.yaml` 文件,通过[下载页面](https://dolphinscheduler.apache.org/en-us/download/<version>)
确保 docker-compose 顺利安装后,需要获取 `docker-compose.yaml` 文件,通过[下载页面](https://dolphinscheduler.apache.org/en-us/download/3.2.1)
下载对应版本源码包可能是最快的方法,当下载完源码后就可以运行命令进行部署了。
```shell
$ DOLPHINSCHEDULER_VERSION=<version>
$ DOLPHINSCHEDULER_VERSION=3.2.1
$ tar -zxf apache-dolphinscheduler-"${DOLPHINSCHEDULER_VERSION}"-src.tar.gz
# Mac Linux 用户
$ cd apache-dolphinscheduler-"${DOLPHINSCHEDULER_VERSION}"-src/deploy/docker
@ -61,7 +61,7 @@ $ docker-compose --profile all up -d
ZooKeeper 且不想启动新的服务,可以使用这个方式分别启动 DolphinScheduler 容器。
```shell
$ DOLPHINSCHEDULER_VERSION=<version>
$ DOLPHINSCHEDULER_VERSION=3.2.1
# 初始化数据库,其确保数据库 <DATABASE> 已经存在
$ docker run -d --name dolphinscheduler-tools \
-e DATABASE="postgresql" \
@ -74,7 +74,7 @@ $ docker run -d --name dolphinscheduler-tools \
# 启动 DolphinScheduler 对应的服务
$ docker run -d --name dolphinscheduler-master \
-e DATABASE="postgresql" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/dolphinscheduler" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/<DATABASE>" \
-e SPRING_DATASOURCE_USERNAME="<USER>" \
-e SPRING_DATASOURCE_PASSWORD="<PASSWORD>" \
-e SPRING_JACKSON_TIME_ZONE="UTC" \
@ -83,7 +83,7 @@ $ docker run -d --name dolphinscheduler-master \
-d apache/dolphinscheduler-master:"${DOLPHINSCHEDULER_VERSION}"
$ docker run -d --name dolphinscheduler-worker \
-e DATABASE="postgresql" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/dolphinscheduler" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/<DATABASE>" \
-e SPRING_DATASOURCE_USERNAME="<USER>" \
-e SPRING_DATASOURCE_PASSWORD="<PASSWORD>" \
-e SPRING_JACKSON_TIME_ZONE="UTC" \
@ -92,7 +92,7 @@ $ docker run -d --name dolphinscheduler-worker \
-d apache/dolphinscheduler-worker:"${DOLPHINSCHEDULER_VERSION}"
$ docker run -d --name dolphinscheduler-api \
-e DATABASE="postgresql" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/dolphinscheduler" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/<DATABASE>" \
-e SPRING_DATASOURCE_USERNAME="<USER>" \
-e SPRING_DATASOURCE_PASSWORD="<PASSWORD>" \
-e SPRING_JACKSON_TIME_ZONE="UTC" \
@ -101,7 +101,7 @@ $ docker run -d --name dolphinscheduler-api \
-d apache/dolphinscheduler-api:"${DOLPHINSCHEDULER_VERSION}"
$ docker run -d --name dolphinscheduler-alert-server \
-e DATABASE="postgresql" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/dolphinscheduler" \
-e SPRING_DATASOURCE_URL="jdbc:postgresql://localhost:5432/<DATABASE>" \
-e SPRING_DATASOURCE_USERNAME="<USER>" \
-e SPRING_DATASOURCE_PASSWORD="<PASSWORD>" \
-e SPRING_JACKSON_TIME_ZONE="UTC" \

View File

@ -4,6 +4,10 @@
#### 以下是Apache DolphinScheduler每个稳定版本的设置说明。
### Versions: 3.2.1
#### Links [3.2.1 Document](../3.2.1/user_doc/about/introduction.md)
### Versions: 3.2.0
#### Links [3.2.0 Document](../3.2.0/user_doc/about/introduction.md)

View File

@ -41,6 +41,12 @@
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-dao</artifactId>
<exclusions>
<exclusion>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-storage-api</artifactId>
</exclusion>
</exclusions>
</dependency>
<dependency>

View File

@ -24,6 +24,7 @@
-XX:+PrintGCDetails
-Xloggc:gc.log
-XX:+ExitOnOutOfMemoryError
-XX:+HeapDumpOnOutOfMemoryError
-XX:HeapDumpPath=dump.hprof

View File

@ -46,5 +46,15 @@
<artifactId>dolphinscheduler-api</artifactId>
<version>dev-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpclient</artifactId>
<version>4.5.13</version>
</dependency>
<dependency>
<groupId>org.apache.httpcomponents</groupId>
<artifactId>httpmime</artifactId>
<version>4.5.13</version>
</dependency>
</dependencies>
</project>

View File

@ -84,8 +84,10 @@ datasource.encryption.enable=false
# datasource encryption salt
datasource.encryption.salt=!@#$%^&*
# data quality option
data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
# data quality jar directory path, it would auto discovery data quality jar from this given dir. You should keep it empty if you do not change anything in
# data-quality, it will auto discovery by dolphinscheduler itself. Change it only if you want to use your own data-quality jar and it is not in worker-server
# libs directory(but may sure your jar name start with `dolphinscheduler-data-quality`).
data-quality.jar.dir=
#data-quality.error.output.path=/tmp/data-quality-error-data

View File

@ -69,7 +69,7 @@
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-task-all</artifactId>
<artifactId>dolphinscheduler-task-all-prune</artifactId>
</dependency>
<dependency>

View File

@ -24,6 +24,7 @@
-XX:+PrintGCDetails
-Xloggc:gc.log
-XX:+ExitOnOutOfMemoryError
-XX:+HeapDumpOnOutOfMemoryError
-XX:HeapDumpPath=dump.hprof

View File

@ -18,10 +18,12 @@
package org.apache.dolphinscheduler.api.controller;
import static org.apache.dolphinscheduler.api.enums.Status.BATCH_EXECUTE_PROCESS_INSTANCE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.BATCH_START_PROCESS_INSTANCE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.CHECK_PROCESS_DEFINITION_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.EXECUTE_PROCESS_INSTANCE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.QUERY_EXECUTING_WORKFLOW_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.START_PROCESS_INSTANCE_ERROR;
import static org.apache.dolphinscheduler.api.enums.Status.START_TASK_INSTANCE_ERROR;
import org.apache.dolphinscheduler.api.enums.ExecuteType;
import org.apache.dolphinscheduler.api.enums.Status;
@ -226,7 +228,7 @@ public class ExecutorController extends BaseController {
})
@PostMapping(value = "batch-start-process-instance")
@ResponseStatus(HttpStatus.OK)
@ApiException(START_PROCESS_INSTANCE_ERROR)
@ApiException(BATCH_START_PROCESS_INSTANCE_ERROR)
public Result batchStartProcessInstance(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@Parameter(name = "projectCode", description = "PROJECT_CODE", required = true) @PathVariable long projectCode,
@RequestParam(value = "processDefinitionCodes") String processDefinitionCodes,
@ -292,7 +294,7 @@ public class ExecutorController extends BaseController {
}
if (!startFailedProcessDefinitionCodeList.isEmpty()) {
putMsg(result, Status.BATCH_START_PROCESS_INSTANCE_ERROR,
putMsg(result, BATCH_START_PROCESS_INSTANCE_ERROR,
String.join(Constants.COMMA, startFailedProcessDefinitionCodeList));
}
@ -437,7 +439,7 @@ public class ExecutorController extends BaseController {
})
@PostMapping(value = "/task-instance/{code}/start")
@ResponseStatus(HttpStatus.OK)
@ApiException(START_PROCESS_INSTANCE_ERROR)
@ApiException(START_TASK_INSTANCE_ERROR)
public Result<Boolean> startStreamTaskInstance(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@Parameter(name = "projectCode", description = "PROJECT_CODE", required = true) @PathVariable long projectCode,
@Parameter(name = "code", description = "TASK_CODE", required = true) @PathVariable long code,

View File

@ -92,7 +92,7 @@ public class K8sNamespaceController extends BaseController {
}
/**
* create namespace,if not exist on k8s,will create,if exist only register in db
* register namespace in db,need to create namespace in k8s first
*
* @param loginUser
* @param namespace k8s namespace
@ -111,7 +111,7 @@ public class K8sNamespaceController extends BaseController {
@RequestParam(value = "namespace") String namespace,
@RequestParam(value = "clusterCode") Long clusterCode) {
Map<String, Object> result =
k8sNamespaceService.createK8sNamespace(loginUser, namespace, clusterCode);
k8sNamespaceService.registerK8sNamespace(loginUser, namespace, clusterCode);
return returnDataList(result);
}

View File

@ -298,7 +298,7 @@ public class TaskGroupController extends BaseController {
* @param pageSize page size
* @return queue list
*/
@Operation(summary = "queryTasksByGroupId", description = "QUERY_ALL_TASKS_GROUP_NOTES")
@Operation(summary = "queryTaskGroupQueuesByGroupId", description = "QUERY_TASKS_GROUP_GROUP_QUEUES")
@Parameters({
@Parameter(name = "groupId", description = "GROUP_ID", required = false, schema = @Schema(implementation = int.class, example = "1", defaultValue = "-1")),
@Parameter(name = "taskInstanceName", description = "TASK_INSTANCE_NAME", required = false, schema = @Schema(implementation = String.class, example = "taskName")),
@ -310,15 +310,21 @@ public class TaskGroupController extends BaseController {
@GetMapping(value = "/query-list-by-group-id")
@ResponseStatus(HttpStatus.OK)
@ApiException(QUERY_TASK_GROUP_QUEUE_LIST_ERROR)
public Result queryTasksByGroupId(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@RequestParam(value = "groupId", required = false, defaultValue = "-1") Integer groupId,
@RequestParam(value = "taskInstanceName", required = false) String taskName,
@RequestParam(value = "processInstanceName", required = false) String processName,
@RequestParam(value = "status", required = false) Integer status,
@RequestParam("pageNo") Integer pageNo,
@RequestParam("pageSize") Integer pageSize) {
Map<String, Object> result = taskGroupQueueService.queryTasksByGroupId(loginUser, taskName, processName, status,
groupId, pageNo, pageSize);
public Result queryTaskGroupQueues(@Parameter(hidden = true) @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@RequestParam(value = "groupId", required = false, defaultValue = "-1") Integer groupId,
@RequestParam(value = "taskInstanceName", required = false) String taskName,
@RequestParam(value = "processInstanceName", required = false) String processName,
@RequestParam(value = "status", required = false) Integer status,
@RequestParam("pageNo") Integer pageNo,
@RequestParam("pageSize") Integer pageSize) {
Map<String, Object> result = taskGroupQueueService.queryTasksByGroupId(
loginUser,
taskName,
processName,
status,
groupId,
pageNo,
pageSize);
return returnDataList(result);
}

View File

@ -41,14 +41,14 @@ public interface K8sNamespaceService {
Result queryListPaging(User loginUser, String searchVal, Integer pageNo, Integer pageSize);
/**
* create namespace,if not exist on k8s,will create,if exist only register in db
* register namespace in db,need to create namespace in k8s first
*
* @param loginUser login user
* @param namespace namespace
* @param clusterCode k8s not null
* @return
*/
Map<String, Object> createK8sNamespace(User loginUser, String namespace, Long clusterCode);
Map<String, Object> registerK8sNamespace(User loginUser, String namespace, Long clusterCode);
/**
* verify namespace and k8s

View File

@ -50,14 +50,6 @@ public interface TaskGroupService {
Map<String, Object> updateTaskGroup(User loginUser, int id, String name,
String description, int groupSize);
/**
* get task group status
*
* @param id task group id
* @return the result code and msg
*/
boolean isTheTaskGroupAvailable(int id);
/**
* query all task group by user id
*

View File

@ -55,7 +55,6 @@ import org.apache.dolphinscheduler.common.enums.Priority;
import org.apache.dolphinscheduler.common.enums.ReleaseState;
import org.apache.dolphinscheduler.common.enums.RunMode;
import org.apache.dolphinscheduler.common.enums.TaskDependType;
import org.apache.dolphinscheduler.common.enums.TaskGroupQueueStatus;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.enums.WorkflowExecutionStatus;
import org.apache.dolphinscheduler.common.model.Server;
@ -83,14 +82,12 @@ import org.apache.dolphinscheduler.dao.mapper.TaskGroupQueueMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.repository.ProcessInstanceDao;
import org.apache.dolphinscheduler.extract.base.client.SingletonJdkDynamicRpcClientProxyFactory;
import org.apache.dolphinscheduler.extract.master.ILogicTaskInstanceOperator;
import org.apache.dolphinscheduler.extract.master.IStreamingTaskOperator;
import org.apache.dolphinscheduler.extract.master.ITaskInstanceExecutionEventListener;
import org.apache.dolphinscheduler.extract.master.IWorkflowInstanceService;
import org.apache.dolphinscheduler.extract.master.dto.WorkflowExecuteDto;
import org.apache.dolphinscheduler.extract.master.transportor.StreamingTaskTriggerRequest;
import org.apache.dolphinscheduler.extract.master.transportor.StreamingTaskTriggerResponse;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceForceStartRequest;
import org.apache.dolphinscheduler.extract.master.transportor.WorkflowInstanceStateChangeEvent;
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
import org.apache.dolphinscheduler.service.command.CommandService;
@ -552,16 +549,20 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
Map<String, Object> result = new HashMap<>();
TaskGroupQueue taskGroupQueue = taskGroupQueueMapper.selectById(queueId);
// check process instance exist
ProcessInstance processInstance = processInstanceMapper.selectById(taskGroupQueue.getProcessId());
if (processInstance == null) {
log.error("Process instance does not exist, projectCode:{}, processInstanceId:{}.",
taskGroupQueue.getProjectCode(), taskGroupQueue.getProcessId());
putMsg(result, Status.PROCESS_INSTANCE_NOT_EXIST, taskGroupQueue.getProcessId());
return result;
}
ProcessInstance processInstance = processInstanceDao.queryOptionalById(taskGroupQueue.getProcessId())
.orElseThrow(
() -> new ServiceException(Status.PROCESS_INSTANCE_NOT_EXIST, taskGroupQueue.getProcessId()));
checkMasterExists();
return forceStart(processInstance, taskGroupQueue);
if (taskGroupQueue.getInQueue() == Flag.NO.getCode()) {
throw new ServiceException(Status.TASK_GROUP_QUEUE_ALREADY_START);
}
taskGroupQueue.setForceStart(Flag.YES.getCode());
taskGroupQueue.setUpdateTime(new Date());
taskGroupQueueMapper.updateById(taskGroupQueue);
result.put(Constants.STATUS, Status.SUCCESS);
return result;
}
public void checkStartNodeList(String startNodeList, Long processDefinitionCode, int version) {
@ -664,32 +665,6 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
return result;
}
/**
* prepare to update process instance command type and status
*
* @param processInstance process instance
* @return update result
*/
private Map<String, Object> forceStart(ProcessInstance processInstance, TaskGroupQueue taskGroupQueue) {
Map<String, Object> result = new HashMap<>();
if (taskGroupQueue.getStatus() != TaskGroupQueueStatus.WAIT_QUEUE) {
log.warn("Task group queue already starts, taskGroupQueueId:{}.", taskGroupQueue.getId());
putMsg(result, Status.TASK_GROUP_QUEUE_ALREADY_START);
return result;
}
taskGroupQueue.setForceStart(Flag.YES.getCode());
taskGroupQueue.setUpdateTime(new Date());
processService.updateTaskGroupQueue(taskGroupQueue);
log.info("Sending force start command to master: {}.", processInstance.getHost());
ILogicTaskInstanceOperator iLogicTaskInstanceOperator = SingletonJdkDynamicRpcClientProxyFactory
.getProxyClient(processInstance.getHost(), ILogicTaskInstanceOperator.class);
iLogicTaskInstanceOperator.forceStartTaskInstance(
new TaskInstanceForceStartRequest(processInstance.getId(), taskGroupQueue.getTaskId()));
putMsg(result, Status.SUCCESS);
return result;
}
/**
* check whether sub processes are offline before starting process definition
*

View File

@ -100,7 +100,7 @@ public class K8SNamespaceServiceImpl extends BaseServiceImpl implements K8sNames
}
/**
* create namespace,if not exist on k8s,will create,if exist only register in db
* register namespace in db,need to create namespace in k8s first
*
* @param loginUser login user
* @param namespace namespace
@ -108,7 +108,7 @@ public class K8SNamespaceServiceImpl extends BaseServiceImpl implements K8sNames
* @return
*/
@Override
public Map<String, Object> createK8sNamespace(User loginUser, String namespace, Long clusterCode) {
public Map<String, Object> registerK8sNamespace(User loginUser, String namespace, Long clusterCode) {
Map<String, Object> result = new HashMap<>();
if (isNotAdmin(loginUser)) {
throw new ServiceException(Status.USER_NO_OPERATION_PERM);

View File

@ -66,8 +66,13 @@ public class TaskGroupQueueServiceImpl extends BaseServiceImpl implements TaskGr
* @return tasks list
*/
@Override
public Map<String, Object> queryTasksByGroupId(User loginUser, String taskName, String processName, Integer status,
int groupId, int pageNo, int pageSize) {
public Map<String, Object> queryTasksByGroupId(User loginUser,
String taskName,
String processName,
Integer status,
int groupId,
int pageNo,
int pageSize) {
Map<String, Object> result = new HashMap<>();
Page<TaskGroupQueue> page = new Page<>(pageNo, pageSize);
PageInfo<TaskGroupQueue> pageInfo = new PageInfo<>(pageNo, pageSize);
@ -79,8 +84,13 @@ public class TaskGroupQueueServiceImpl extends BaseServiceImpl implements TaskGr
return result;
}
List<Project> projects = projectMapper.selectBatchIds(projectIds);
IPage<TaskGroupQueue> taskGroupQueue = taskGroupQueueMapper.queryTaskGroupQueueByTaskGroupIdPaging(page,
taskName, processName, status, groupId, projects);
IPage<TaskGroupQueue> taskGroupQueue = taskGroupQueueMapper.queryTaskGroupQueueByTaskGroupIdPaging(
page,
taskName,
processName,
status,
groupId,
projects);
pageInfo.setTotal((int) taskGroupQueue.getTotal());
pageInfo.setTotalList(taskGroupQueue.getRecords());

View File

@ -122,7 +122,7 @@ public class TaskGroupServiceImpl extends BaseServiceImpl implements TaskGroupSe
.description(description)
.groupSize(groupSize)
.userId(loginUser.getId())
.status(Flag.YES.getCode())
.status(Flag.YES)
.createTime(now)
.updateTime(now)
.build();
@ -180,7 +180,7 @@ public class TaskGroupServiceImpl extends BaseServiceImpl implements TaskGroupSe
putMsg(result, Status.TASK_GROUP_NAME_EXSIT);
return result;
}
if (taskGroup.getStatus() != Flag.YES.getCode()) {
if (taskGroup.getStatus() != Flag.YES) {
log.warn("Task group has been closed, taskGroupId:{}.", id);
putMsg(result, Status.TASK_GROUP_STATUS_ERROR);
return result;
@ -202,17 +202,6 @@ public class TaskGroupServiceImpl extends BaseServiceImpl implements TaskGroupSe
return result;
}
/**
* get task group status
*
* @param id task group id
* @return is the task group available
*/
@Override
public boolean isTheTaskGroupAvailable(int id) {
return taskGroupMapper.selectCountByIdStatus(id, Flag.YES.getCode()) == 1;
}
/**
* query all task group by user id
*
@ -331,12 +320,12 @@ public class TaskGroupServiceImpl extends BaseServiceImpl implements TaskGroupSe
return result;
}
TaskGroup taskGroup = taskGroupMapper.selectById(id);
if (taskGroup.getStatus() == Flag.NO.getCode()) {
if (taskGroup.getStatus() == Flag.NO) {
log.info("Task group has been closed, taskGroupId:{}.", id);
putMsg(result, Status.TASK_GROUP_STATUS_CLOSED);
return result;
}
taskGroup.setStatus(Flag.NO.getCode());
taskGroup.setStatus(Flag.NO);
int update = taskGroupMapper.updateById(taskGroup);
if (update > 0)
log.info("Task group close complete, taskGroupId:{}.", id);
@ -364,12 +353,12 @@ public class TaskGroupServiceImpl extends BaseServiceImpl implements TaskGroupSe
return result;
}
TaskGroup taskGroup = taskGroupMapper.selectById(id);
if (taskGroup.getStatus() == Flag.YES.getCode()) {
if (taskGroup.getStatus() == Flag.YES) {
log.info("Task group has been started, taskGroupId:{}.", id);
putMsg(result, Status.TASK_GROUP_STATUS_OPENED);
return result;
}
taskGroup.setStatus(Flag.YES.getCode());
taskGroup.setStatus(Flag.YES);
taskGroup.setUpdateTime(new Date(System.currentTimeMillis()));
int update = taskGroupMapper.updateById(taskGroup);
if (update > 0)

View File

@ -255,6 +255,7 @@ public class TaskInstanceServiceImpl extends BaseServiceImpl implements TaskInst
// change the state of the task instance
task.setState(TaskExecutionStatus.FORCED_SUCCESS);
task.setEndTime(new Date());
int changedNum = taskInstanceMapper.updateById(task);
if (changedNum > 0) {
processService.forceProcessInstanceSuccessByTaskInstanceId(taskInstanceId);

View File

@ -66,7 +66,6 @@ import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.facebook.presto.jdbc.internal.guava.base.Strings;
/**
* worker group service impl
@ -247,7 +246,7 @@ public class WorkerGroupServiceImpl extends BaseServiceImpl implements WorkerGro
* @return boolean
*/
private String checkWorkerGroupAddrList(WorkerGroup workerGroup) {
if (Strings.isNullOrEmpty(workerGroup.getAddrList())) {
if (StringUtils.isEmpty(workerGroup.getAddrList())) {
return null;
}
Map<String, String> serverMaps = registryClient.getServerMaps(RegistryNodeType.WORKER);

View File

@ -107,16 +107,16 @@ public class K8SNamespaceServiceTest {
public void createK8sNamespace() {
// namespace is null
Map<String, Object> result =
k8sNamespaceService.createK8sNamespace(getLoginUser(), null, clusterCode);
k8sNamespaceService.registerK8sNamespace(getLoginUser(), null, clusterCode);
logger.info(result.toString());
Assertions.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR, result.get(Constants.STATUS));
// k8s is null
result = k8sNamespaceService.createK8sNamespace(getLoginUser(), namespace, null);
result = k8sNamespaceService.registerK8sNamespace(getLoginUser(), namespace, null);
logger.info(result.toString());
Assertions.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR, result.get(Constants.STATUS));
// correct
Mockito.when(clusterMapper.queryByClusterCode(Mockito.anyLong())).thenReturn(getCluster());
result = k8sNamespaceService.createK8sNamespace(getLoginUser(), namespace, clusterCode);
result = k8sNamespaceService.registerK8sNamespace(getLoginUser(), namespace, clusterCode);
logger.info(result.toString());
Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}

View File

@ -769,7 +769,6 @@ public class ProcessInstanceServiceTest {
processInstance.getProcessDefinitionCode(),
processInstance.getProcessDefinitionVersion())).thenReturn(new ProcessDefinitionLog());
when(processInstanceMapper.queryDetailById(1)).thenReturn(processInstance);
when(taskInstanceMapper.queryByInstanceIdAndName(Mockito.anyInt(), Mockito.any())).thenReturn(taskInstance);
DAG<Long, TaskNode, TaskNodeRelation> graph = new DAG<>();
for (long i = 1; i <= 7; ++i) {
graph.addNode(i, new TaskNode());

View File

@ -125,7 +125,7 @@ public class TaskGroupServiceTest {
.description(taskGroupDesc)
.groupSize(100)
.userId(1)
.status(Flag.YES.getCode())
.status(Flag.YES)
.build();
return taskGroup;
@ -204,7 +204,7 @@ public class TaskGroupServiceTest {
User loginUser = getLoginUser();
TaskGroup taskGroup = getTaskGroup();
taskGroup.setStatus(Flag.YES.getCode());
taskGroup.setStatus(Flag.YES);
// Task group status error
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.TASK_GROUP,
@ -218,7 +218,6 @@ public class TaskGroupServiceTest {
logger.info(result.toString());
Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
taskGroup.setStatus(0);
}
@Test
@ -236,12 +235,12 @@ public class TaskGroupServiceTest {
Map<String, Object> result = taskGroupService.closeTaskGroup(loginUser, 1);
Assertions.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
taskGroup.setStatus(0);
taskGroup.setStatus(Flag.NO);
Mockito.when(taskGroupMapper.selectById(1)).thenReturn(taskGroup);
result = taskGroupService.closeTaskGroup(loginUser, 1);
Assertions.assertEquals(Status.TASK_GROUP_STATUS_CLOSED, result.get(Constants.STATUS));
taskGroup.setStatus(1);
taskGroup.setStatus(Flag.YES);
Mockito.when(taskGroupMapper.selectById(1)).thenReturn(taskGroup);
result = taskGroupService.startTaskGroup(loginUser, 1);
Assertions.assertEquals(Status.TASK_GROUP_STATUS_OPENED, result.get(Constants.STATUS));

View File

@ -102,7 +102,7 @@ public class DataSourceConstants {
* dataSource sensitive param
*/
public static final String DATASOURCE_PASSWORD_REGEX =
"(?<=((?i)password((\":\")|(\\\\\":\\\\\")|(=')))).*?(?=((\")|(\\\\\")|(')))";
"(?<=((?i)password((\" : \")|(\":\")|(\\\\\":\\\\\")|(=')))).*?(?=((\")|(\\\\\")|(')))";
/**
* datasource encryption salt

View File

@ -25,8 +25,6 @@ import static org.apache.dolphinscheduler.common.constants.Constants.RESOURCE_VI
import static org.apache.dolphinscheduler.common.constants.Constants.UTF_8;
import static org.apache.dolphinscheduler.common.constants.DateConstants.YYYYMMDDHHMMSS;
import org.apache.dolphinscheduler.common.exception.FileOperateException;
import org.apache.commons.io.IOUtils;
import org.apache.commons.lang3.SystemUtils;
@ -37,10 +35,9 @@ import java.io.FileOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.charset.StandardCharsets;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.Path;
import java.nio.file.attribute.FileAttribute;
import java.nio.file.attribute.PosixFilePermission;
import java.nio.file.attribute.PosixFilePermissions;
import java.util.Set;
@ -48,11 +45,10 @@ import java.util.zip.CRC32;
import java.util.zip.CheckedInputStream;
import lombok.NonNull;
import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;
/**
* file utils
*/
@UtilityClass
@Slf4j
public class FileUtils {
@ -62,14 +58,7 @@ public class FileUtils {
public static final String KUBE_CONFIG_FILE = "config";
private static final String RWXR_XR_X = "rwxr-xr-x";
private static final FileAttribute<Set<PosixFilePermission>> PERMISSION_755 =
PosixFilePermissions.asFileAttribute(PosixFilePermissions.fromString(RWXR_XR_X));
private FileUtils() {
throw new UnsupportedOperationException("Construct FileUtils");
}
private static final Set<PosixFilePermission> PERMISSION_755 = PosixFilePermissions.fromString("rwxr-xr-x");
/**
* get download file absolute path and name
@ -161,34 +150,6 @@ public class FileUtils {
return PropertyUtils.getString(RESOURCE_VIEW_SUFFIXES, RESOURCE_VIEW_SUFFIXES_DEFAULT_VALUE);
}
/**
* create directory if absent
*
* @param execLocalPath execute local path
* @throws IOException errors
*/
public static void createWorkDirIfAbsent(String execLocalPath) throws IOException {
// if work dir exists, first delete
File execLocalPathFile = new File(execLocalPath);
if (execLocalPathFile.exists()) {
try {
org.apache.commons.io.FileUtils.forceDelete(execLocalPathFile);
} catch (Exception ex) {
if (ex instanceof NoSuchFileException || ex.getCause() instanceof NoSuchFileException) {
// this file is already be deleted.
} else {
throw ex;
}
}
}
// create work dir
org.apache.commons.io.FileUtils.forceMkdir(execLocalPathFile);
String mkdirLog = "create dir success " + execLocalPath;
log.info(mkdirLog);
}
/**
* write content to file ,if parent path not exists, it will do one's utmost to mkdir
*
@ -231,25 +192,6 @@ public class FileUtils {
org.apache.commons.io.FileUtils.deleteQuietly(new File(filename));
}
/**
* Gets all the parent subdirectories of the parentDir directory
*
* @param parentDir parent dir
* @return all dirs
*/
public static File[] getAllDir(String parentDir) {
if (parentDir == null || "".equals(parentDir)) {
throw new RuntimeException("parentDir can not be empty");
}
File file = new File(parentDir);
if (!file.exists() || !file.isDirectory()) {
throw new RuntimeException("parentDir not exist, or is not a directory:" + parentDir);
}
return file.listFiles(File::isDirectory);
}
/**
* Get Content
*
@ -325,59 +267,52 @@ public class FileUtils {
return crcString;
}
public static void setFileOwner(Path filePath, String fileOwner) throws FileOperateException {
try {
// We use linux command to set the file owner, since jdk api will not use sudo.
String command = String.format("sudo chown %s %s", fileOwner, filePath.toString());
Runtime.getRuntime().exec(command);
Process process = Runtime.getRuntime().exec(command);
int exitCode = process.waitFor();
if (0 != exitCode) {
throw new FileOperateException(
"Set file: " + filePath + " to owner: " + fileOwner + " failed, existCode(" + exitCode + ")");
}
} catch (FileOperateException ex) {
throw ex;
} catch (Exception ex) {
throw new FileOperateException("Set directory: " + filePath + " to owner: " + fileOwner + " failed");
}
}
public static void setDirectoryOwner(Path filePath, String fileOwner) throws FileOperateException {
try {
// We use linux command to set the file owner, since jdk api will not use sudo.
String command = String.format("sudo chown -R %s %s", fileOwner, filePath.toString());
Runtime.getRuntime().exec(command);
Process process = Runtime.getRuntime().exec(command);
int exitCode = process.waitFor();
if (0 != exitCode) {
throw new FileOperateException("Set directory: " + filePath + " to owner: " + fileOwner
+ " failed, existCode(" + exitCode + ")");
}
} catch (FileOperateException ex) {
throw ex;
} catch (Exception ex) {
throw new FileOperateException("Set directory: " + filePath + " to owner: " + fileOwner + " failed");
}
}
public static void createDirectoryIfNotPresent(Path path) throws IOException {
if (Files.exists(path)) {
return;
}
Files.createDirectories(path);
}
/**
* Create a file with '755'.
*/
public static void createFileWith755(@NonNull Path path) throws IOException {
if (SystemUtils.IS_OS_WINDOWS) {
Files.createFile(path);
} else {
Files.createFile(path, PERMISSION_755);
Files.createFile(path);
Files.setPosixFilePermissions(path, PERMISSION_755);
}
}
public static void createDirectoryWith755(@NonNull Path path) throws IOException {
if (path.toFile().exists()) {
return;
}
if (OSUtils.isWindows()) {
Files.createDirectories(path);
} else {
Path parent = path.getParent();
if (parent != null && !parent.toFile().exists()) {
createDirectoryWith755(parent);
}
try {
Files.createDirectory(path);
Files.setPosixFilePermissions(path, PERMISSION_755);
} catch (FileAlreadyExistsException fileAlreadyExistsException) {
// Catch the FileAlreadyExistsException here to avoid create the same parent directory in parallel
log.debug("The directory: {} already exists", path);
}
}
}
public static void setFileTo755(File file) throws IOException {
if (OSUtils.isWindows()) {
return;
}
if (file.isFile()) {
Files.setPosixFilePermissions(file.toPath(), PERMISSION_755);
return;
}
Files.setPosixFilePermissions(file.toPath(), PERMISSION_755);
File[] files = file.listFiles();
if (files != null) {
for (File f : files) {
setFileTo755(f);
}
}
}

View File

@ -120,8 +120,10 @@ datasource.encryption.enable=false
# datasource encryption salt
datasource.encryption.salt=!@#$%^&*
# data quality option
data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
# data quality jar directory path, it would auto discovery data quality jar from this given dir. You should keep it empty if you do not change anything in
# data-quality, it will auto discovery by dolphinscheduler itself. Change it only if you want to use your own data-quality jar and it is not in worker-server
# libs directory(but may sure your jar name start with `dolphinscheduler-data-quality`).
data-quality.jar.dir=
#data-quality.error.output.path=/tmp/data-quality-error-data

View File

@ -79,6 +79,34 @@ public class SensitiveDataConverterTest {
" }\n" +
"}");
// data quality
tcs.put("\"readers\" : [ {\n" +
" \"type\" : \"JDBC\",\n" +
" \"config\" : {\n" +
" \"database\" : \"dolphinscheduler\",\n" +
" \"password\" : \"view1\",\n" +
" \"driver\" : \"com.mysql.cj.jdbc.Driver\",\n" +
" \"user\" : \"root\",\n" +
" \"output_table\" : \"dolphinscheduler_users\",\n" +
" \"table\" : \"users\",\n" +
" \"url\" : \"jdbc:mysql://127.0.0.1:3307/dolphinscheduler?userSSL=true&enabledTLSProtocols=TLSv1.2\"\n"
+
" }\n" +
" } ]",
"\"readers\" : [ {\n" +
" \"type\" : \"JDBC\",\n" +
" \"config\" : {\n" +
" \"database\" : \"dolphinscheduler\",\n" +
" \"password\" : \"*****\",\n" +
" \"driver\" : \"com.mysql.cj.jdbc.Driver\",\n" +
" \"user\" : \"root\",\n" +
" \"output_table\" : \"dolphinscheduler_users\",\n" +
" \"table\" : \"users\",\n" +
" \"url\" : \"jdbc:mysql://127.0.0.1:3307/dolphinscheduler?userSSL=true&enabledTLSProtocols=TLSv1.2\"\n"
+
" }\n" +
" } ]");
for (String logMsg : tcs.keySet()) {
String maskedLog = SensitiveDataConverter.maskSensitiveData(logMsg);
logger.info("original parameter : {}", logMsg);

View File

@ -19,8 +19,13 @@ package org.apache.dolphinscheduler.common.utils;
import static org.apache.dolphinscheduler.common.constants.DateConstants.YYYYMMDDHHMMSS;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
@ -55,12 +60,23 @@ public class FileUtilsTest {
}
@Test
public void testCreateWorkDirIfAbsent() {
public void createDirectoryWith755() throws IOException {
Path path = Paths.get("/tmp/createWorkDirAndUserIfAbsent");
try {
FileUtils.createWorkDirIfAbsent("/tmp/createWorkDirAndUserIfAbsent");
Assertions.assertTrue(true);
FileUtils.createDirectoryWith755(path);
File file = path.toFile();
Assertions.assertTrue(file.exists());
Assertions.assertTrue(file.isDirectory());
Assertions.assertTrue(file.canExecute());
Assertions.assertTrue(file.canRead());
Assertions.assertTrue(file.canWrite());
FileUtils.createDirectoryWith755(Paths.get("/"));
} catch (Exception e) {
Assertions.fail();
e.printStackTrace();
Assertions.fail(e.getMessage());
} finally {
Files.deleteIfExists(path);
}
}

View File

@ -115,8 +115,10 @@ datasource.encryption.enable=false
# datasource encryption salt
datasource.encryption.salt=!@#$%^&*
# data quality option
data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
# data quality jar directory path, it would auto discovery data quality jar from this given dir. You should keep it empty if you do not change anything in
# data-quality, it will auto discovery by dolphinscheduler itself. Change it only if you want to use your own data-quality jar and it is not in worker-server
# libs directory(but may sure your jar name start with `dolphinscheduler-data-quality`).
data-quality.jar.dir=
#data-quality.error.output.path=/tmp/data-quality-error-data

View File

@ -17,6 +17,8 @@
package org.apache.dolphinscheduler.dao.entity;
import org.apache.dolphinscheduler.common.enums.Flag;
import java.io.Serializable;
import java.util.Date;
@ -36,44 +38,17 @@ import com.baomidou.mybatisplus.annotation.TableName;
@TableName("t_ds_task_group")
public class TaskGroup implements Serializable {
/**
* key
*/
@TableId(value = "id", type = IdType.AUTO)
private Integer id;
/**
* task_group name
*/
private String name;
private String description;
/**
* 作业组大小
*/
private int groupSize;
/**
* 已使用作业组大小
*/
private int useSize;
/**
* creator id
*/
private int userId;
/**
* 0 not available, 1 available
*/
private Integer status;
/**
* create time
*/
private Date createTime;
/**
* update time
*/
private Date updateTime;
/**
* project code
*/
private long projectCode;
private String description;
private int groupSize;
private int useSize;
private int userId;
private Flag status;
private Date createTime;
private Date updateTime;
}

View File

@ -35,20 +35,6 @@ import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
*/
public interface TaskGroupMapper extends BaseMapper<TaskGroup> {
int robTaskGroupResource(@Param("id") int id,
@Param("currentUseSize") int currentUseSize,
@Param("queueId") int queueId,
@Param("queueStatus") int queueStatus);
/**
* update table of task group
*
* @param id primary key
* @return affected rows
*/
int releaseTaskGroupResource(@Param("id") int id, @Param("useSize") int useSize,
@Param("queueId") int queueId, @Param("queueStatus") int queueStatus);
/**
* select task groups paging
*
@ -57,7 +43,8 @@ public interface TaskGroupMapper extends BaseMapper<TaskGroup> {
* @param status status
* @return result page
*/
IPage<TaskGroup> queryTaskGroupPaging(IPage<TaskGroup> page, @Param("name") String name,
IPage<TaskGroup> queryTaskGroupPaging(IPage<TaskGroup> page,
@Param("name") String name,
@Param("status") Integer status);
/**
@ -69,13 +56,6 @@ public interface TaskGroupMapper extends BaseMapper<TaskGroup> {
*/
TaskGroup queryByName(@Param("userId") int userId, @Param("name") String name);
/**
* Select the groupSize > useSize Count
*/
int selectAvailableCountById(@Param("groupId") int groupId);
int selectCountByIdStatus(@Param("id") int id, @Param("status") int status);
IPage<TaskGroup> queryTaskGroupPagingByProjectCode(Page<TaskGroup> page, @Param("projectCode") Long projectCode);
/**
@ -87,4 +67,12 @@ public interface TaskGroupMapper extends BaseMapper<TaskGroup> {
List<TaskGroup> listAuthorizedResource(@Param("userId") int userId);
List<TaskGroup> selectByProjectCode(@Param("projectCode") long projectCode);
List<TaskGroup> queryAvailableTaskGroups();
List<TaskGroup> queryUsedTaskGroups();
int acquireTaskGroupSlot(@Param("id") Integer id);
int releaseTaskGroupSlot(@Param("id") Integer id);
}

View File

@ -104,5 +104,22 @@ public interface TaskGroupQueueMapper extends BaseMapper<TaskGroupQueue> {
void deleteByWorkflowInstanceId(@Param("workflowInstanceId") Integer workflowInstanceId);
void deleteByWorkflowInstanceIds(@Param("workflowInstanceIds") List<Integer> workflowInstanceIds);
void deleteByTaskGroupIds(@Param("taskGroupIds") List<Integer> taskGroupIds);
void updateTaskGroupPriorityByTaskInstanceId(@Param("taskInstanceId") Integer taskInstanceId,
@Param("priority") int taskGroupPriority);
List<TaskGroupQueue> queryAllInQueueTaskGroupQueueByGroupId(@Param("taskGroupId") Integer taskGroupId,
@Param("inQueue") int inQueue);
List<TaskGroupQueue> queryAllTaskGroupQueueByInQueue(@Param("inQueue") int inQueue);
List<TaskGroupQueue> queryByTaskInstanceId(@Param("taskInstanceId") Integer taskInstanceId);
List<TaskGroupQueue> queryUsingTaskGroupQueueByGroupId(@Param("taskGroupId") Integer taskGroupId,
@Param("status") int status,
@Param("inQueue") int inQueue,
@Param("forceStart") int forceStart);
}

View File

@ -39,23 +39,10 @@ import com.baomidou.mybatisplus.core.metadata.IPage;
*/
public interface TaskInstanceMapper extends BaseMapper<TaskInstance> {
List<Integer> queryTaskByProcessIdAndState(@Param("processInstanceId") Integer processInstanceId,
@Param("state") Integer state);
List<TaskInstance> findValidTaskListByProcessId(@Param("processInstanceId") Integer processInstanceId,
@Param("flag") Flag flag,
@Param("testFlag") int testFlag);
List<TaskInstance> queryByHostAndStatus(@Param("host") String host,
@Param("states") int[] stateArray);
int setFailoverByHostAndStateArray(@Param("host") String host,
@Param("states") int[] stateArray,
@Param("destStatus") TaskExecutionStatus destStatus);
TaskInstance queryByInstanceIdAndName(@Param("processInstanceId") int processInstanceId,
@Param("name") String name);
TaskInstance queryByInstanceIdAndCode(@Param("processInstanceId") int processInstanceId,
@Param("taskCode") Long taskCode);
@ -66,9 +53,6 @@ public interface TaskInstanceMapper extends BaseMapper<TaskInstance> {
List<TaskInstance> queryByProcessInstanceIdsAndTaskCodes(@Param("processInstanceIds") List<Integer> processInstanceIds,
@Param("taskCodes") List<Long> taskCodes);
Integer countTask(@Param("projectCodes") Long[] projectCodes,
@Param("taskIds") int[] taskIds);
/**
* Statistics task instance group by given project codes list by start time
* <p>
@ -97,20 +81,6 @@ public interface TaskInstanceMapper extends BaseMapper<TaskInstance> {
@Param("endTime") Date endTime,
@Param("projectIds") Set<Integer> projectIds);
/**
* Statistics task instance group by given project codes list by submit time
* <p>
* We only need project codes to determine whether the task instance belongs to the user or not.
*
* @param startTime Statistics start time
* @param endTime Statistics end time
* @param projectCodes Project codes list to filter
* @return List of ExecuteStatusCount
*/
List<ExecuteStatusCount> countTaskInstanceStateByProjectCodesAndStatesBySubmitTime(@Param("startTime") Date startTime,
@Param("endTime") Date endTime,
@Param("projectCodes") Long[] projectCodes,
@Param("states") List<TaskExecutionStatus> states);
/**
* Statistics task instance group by given project codes list by submit time
* <p>
@ -159,9 +129,6 @@ public interface TaskInstanceMapper extends BaseMapper<TaskInstance> {
@Param("startTime") Date startTime,
@Param("endTime") Date endTime);
List<TaskInstance> loadAllInfosNoRelease(@Param("processInstanceId") int processInstanceId,
@Param("status") int status);
void deleteByWorkflowInstanceId(@Param("workflowInstanceId") int workflowInstanceId);
List<TaskInstance> findByWorkflowInstanceId(@Param("workflowInstanceId") Integer workflowInstanceId);

View File

@ -0,0 +1,62 @@
/*
* 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.TaskGroup;
import java.util.List;
public interface TaskGroupDao extends IDao<TaskGroup> {
/**
* Query all TaskGroups
*
* @return all TaskGroups
*/
List<TaskGroup> queryAllTaskGroups();
/**
* Query all TaskGroups which useSize > 0
*
* @return the TaskGroups which useSize > 0
*/
List<TaskGroup> queryUsedTaskGroups();
/**
* Query all TaskGroups which useSize < groupSize
*
* @return the TaskGroups which useSize < groupSize
*/
List<TaskGroup> queryAvailableTaskGroups();
/**
* Acquire a slot for the TaskGroup which useSize should < groupSize, set the useSize = useSize + 1.
*
* @param taskGroupId taskGroupId which shouldn't be null
* @return true if acquire successfully, false otherwise.
*/
boolean acquireTaskGroupSlot(Integer taskGroupId);
/**
* Release a slot for the TaskGroup which useSize should > 0, set the useSize = useSize - 1.
*
* @param taskGroupId taskGroupId which shouldn't be null
* @return true if release successfully, false otherwise.
*/
boolean releaseTaskGroupSlot(Integer taskGroupId);
}

View File

@ -0,0 +1,64 @@
/*
* 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.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskGroupQueue;
import java.util.List;
public interface TaskGroupQueueDao extends IDao<TaskGroupQueue> {
/**
* Delete {@link TaskGroupQueue} by {@link ProcessInstance#getId()}
*
* @param workflowInstanceIds workflowInstanceIds
*/
void deleteByWorkflowInstanceIds(List<Integer> workflowInstanceIds);
/**
* Query all {@link TaskGroupQueue} which in_queue is {@link org.apache.dolphinscheduler.common.enums.Flag#YES}
*
* @return TaskGroupQueue ordered by priority desc
*/
List<TaskGroupQueue> queryAllInQueueTaskGroupQueue();
/**
* Query all {@link TaskGroupQueue} which in_queue is {@link org.apache.dolphinscheduler.common.enums.Flag#YES} and taskGroupId is taskGroupId
*
* @param taskGroupId taskGroupId
* @return TaskGroupQueue ordered by priority desc
*/
List<TaskGroupQueue> queryAllInQueueTaskGroupQueueByGroupId(Integer taskGroupId);
/**
* Query all {@link TaskGroupQueue} which taskId is taskInstanceId
*
* @param taskInstanceId taskInstanceId
* @return TaskGroupQueue ordered by priority desc
*/
List<TaskGroupQueue> queryByTaskInstanceId(Integer taskInstanceId);
/**
* Query all {@link TaskGroupQueue} which status is TaskGroupQueueStatus.ACQUIRE_SUCCESS and forceStart is {@link org.apache.dolphinscheduler.common.enums.Flag#NO}.
*
* @param taskGroupId taskGroupId
* @return TaskGroupQueue
*/
List<TaskGroupQueue> queryAcquiredTaskGroupQueueByGroupId(Integer taskGroupId);
}

View File

@ -100,9 +100,10 @@ public interface TaskInstanceDao extends IDao<TaskInstance> {
/**
* find last task instance corresponding to taskCode in the date interval
*
* @param processInstanceId Task's parent process instance id
* @param depTaskCode taskCode
* @param testFlag test flag
* @param depTaskCode taskCode
* @param testFlag test flag
* @return task instance
*/
TaskInstance queryLastTaskInstanceIntervalInProcessInstance(Integer processInstanceId,

View File

@ -0,0 +1,70 @@
/*
* 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.TaskGroup;
import org.apache.dolphinscheduler.dao.mapper.TaskGroupMapper;
import org.apache.dolphinscheduler.dao.repository.BaseDao;
import org.apache.dolphinscheduler.dao.repository.TaskGroupDao;
import java.util.List;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Repository;
@Slf4j
@Repository
public class TaskGroupDaoImpl extends BaseDao<TaskGroup, TaskGroupMapper> implements TaskGroupDao {
public TaskGroupDaoImpl(@NonNull TaskGroupMapper taskGroupMapper) {
super(taskGroupMapper);
}
@Override
public List<TaskGroup> queryAllTaskGroups() {
return mybatisMapper.selectList(null);
}
@Override
public List<TaskGroup> queryUsedTaskGroups() {
return mybatisMapper.queryUsedTaskGroups();
}
@Override
public List<TaskGroup> queryAvailableTaskGroups() {
return mybatisMapper.queryAvailableTaskGroups();
}
@Override
public boolean acquireTaskGroupSlot(Integer taskGroupId) {
if (taskGroupId == null) {
throw new IllegalArgumentException("taskGroupId cannot be null");
}
return mybatisMapper.acquireTaskGroupSlot(taskGroupId) > 0;
}
@Override
public boolean releaseTaskGroupSlot(Integer taskGroupId) {
if (taskGroupId == null) {
throw new IllegalArgumentException("taskGroupId cannot be null");
}
return mybatisMapper.releaseTaskGroupSlot(taskGroupId) > 0;
}
}

View File

@ -0,0 +1,73 @@
/*
* 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.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.TaskGroupQueueStatus;
import org.apache.dolphinscheduler.dao.entity.TaskGroupQueue;
import org.apache.dolphinscheduler.dao.mapper.TaskGroupQueueMapper;
import org.apache.dolphinscheduler.dao.repository.BaseDao;
import org.apache.dolphinscheduler.dao.repository.TaskGroupQueueDao;
import org.apache.commons.collections4.CollectionUtils;
import java.util.List;
import lombok.NonNull;
import org.springframework.stereotype.Repository;
@Repository
public class TaskGroupQueueDaoImpl extends BaseDao<TaskGroupQueue, TaskGroupQueueMapper> implements TaskGroupQueueDao {
public TaskGroupQueueDaoImpl(@NonNull TaskGroupQueueMapper taskGroupQueueMapper) {
super(taskGroupQueueMapper);
}
@Override
public void deleteByWorkflowInstanceIds(List<Integer> workflowInstanceIds) {
if (CollectionUtils.isEmpty(workflowInstanceIds)) {
return;
}
mybatisMapper.deleteByWorkflowInstanceIds(workflowInstanceIds);
}
@Override
public List<TaskGroupQueue> queryAllInQueueTaskGroupQueue() {
return mybatisMapper.queryAllTaskGroupQueueByInQueue(Flag.YES.getCode());
}
@Override
public List<TaskGroupQueue> queryAllInQueueTaskGroupQueueByGroupId(Integer taskGroupId) {
return mybatisMapper.queryAllInQueueTaskGroupQueueByGroupId(taskGroupId, Flag.YES.getCode());
}
@Override
public List<TaskGroupQueue> queryByTaskInstanceId(Integer taskInstanceId) {
return mybatisMapper.queryByTaskInstanceId(taskInstanceId);
}
@Override
public List<TaskGroupQueue> queryAcquiredTaskGroupQueueByGroupId(Integer taskGroupId) {
return mybatisMapper.queryUsingTaskGroupQueueByGroupId(
taskGroupId,
TaskGroupQueueStatus.ACQUIRE_SUCCESS.getCode(),
Flag.YES.getCode(),
Flag.NO.getCode());
}
}

View File

@ -67,24 +67,6 @@
where project_code = #{projectCode}
</select>
<!--modify data by id-->
<update id="robTaskGroupResource">
update t_ds_task_group
set use_size = use_size + 1
where id = #{id}
and use_size &lt; group_size
and use_size = #{currentUseSize}
and (select count(1) FROM t_ds_task_group_queue where id = #{queueId} and status = #{queueStatus}) = 1
</update>
<!--modify data by id-->
<update id="releaseTaskGroupResource">
update t_ds_task_group
set use_size = use_size-1
where id = #{id} and use_size > 0 and
(select count(1) FROM t_ds_task_group_queue where id = #{queueId} and status = #{queueStatus} ) = 1
</update>
<select id="queryByName" resultType="org.apache.dolphinscheduler.dao.entity.TaskGroup">
select
<include refid="baseSql" />
@ -93,22 +75,6 @@
user_id = #{userId} and name = #{name}
</select>
<select id="selectAvailableCountById" resultType="java.lang.Integer">
select
count(1)
from t_ds_task_group
where
id = #{groupId} and use_size &lt; group_size
</select>
<select id="selectCountByIdStatus" resultType="java.lang.Integer">
select
count(1)
from t_ds_task_group
where
id = #{id} and status = #{status}
</select>
<select id="listAuthorizedResource" resultType="org.apache.dolphinscheduler.dao.entity.TaskGroup">
select
<include refid="baseSql" />
@ -119,4 +85,32 @@
</if>
</select>
<select id="queryAvailableTaskGroups" resultType="org.apache.dolphinscheduler.dao.entity.TaskGroup">
select
<include refid="baseSql"/>
from t_ds_task_group
where use_size <![CDATA[ < ]]> group_size
</select>
<select id="queryUsedTaskGroups" resultType="org.apache.dolphinscheduler.dao.entity.TaskGroup">
select
<include refid="baseSql"/>
from t_ds_task_group
where use_size <![CDATA[ > ]]> 0
</select>
<update id="acquireTaskGroupSlot">
update t_ds_task_group
set use_size = use_size + 1
where id = #{id}
and use_size &lt; group_size
</update>
<update id="releaseTaskGroupSlot">
update t_ds_task_group
set use_size = use_size - 1
where id = #{id}
and use_size > 0
</update>
</mapper>

View File

@ -46,7 +46,7 @@
and group_id = #{groupId}
</if>
</where>
order by update_time desc
order by update_time desc, id desc
</select>
<select id="queryByStatus" resultMap="TaskGroupQueueMap" resultType="map">
@ -126,9 +126,19 @@
<select id="queryTaskGroupQueueByTaskGroupIdPaging" resultType="org.apache.dolphinscheduler.dao.entity.TaskGroupQueue">
select
queue.id, queue.task_name, queue.group_id, queue.process_id, queue.priority, queue.status
, queue.force_start, queue.create_time, queue.update_time,
process.name as processInstanceName,p.name as projectName,p.code as projectCode
queue.id,
queue.task_name,
queue.group_id,
queue.process_id,
queue.priority,
queue.in_queue,
queue.status,
queue.force_start,
queue.create_time,
queue.update_time,
process.name as processInstanceName,
p.name as projectName,
p.code as projectCode
from t_ds_task_group_queue queue
left join t_ds_process_instance process on queue.process_id = process.id
left join t_ds_process_definition p_f on process.process_definition_code = p_f.code
@ -171,6 +181,15 @@
where process_id = #{workflowInstanceId}
</delete>
<delete id="deleteByWorkflowInstanceIds">
delete
from t_ds_task_group_queue
where process_id in
<foreach collection="workflowInstanceIds" index="index" item="i" open="(" separator="," close=")">
#{i}
</foreach>
</delete>
<delete id="deleteByTaskGroupIds">
delete
from t_ds_task_group_queue
@ -180,4 +199,38 @@
</foreach>
</delete>
<update id="updateTaskGroupPriorityByTaskInstanceId">
update t_ds_task_group_queue
set priority = #{priority}
where task_id = #{taskInstanceId}
</update>
<select id="queryAllInQueueTaskGroupQueueByGroupId" resultType="org.apache.dolphinscheduler.dao.entity.TaskGroupQueue">
select
<include refid="baseSql" />
from t_ds_task_group_queue
where group_id = #{taskGroupId} and in_queue = #{inQueue} order by priority desc
</select>
<select id="queryAllTaskGroupQueueByInQueue" resultType="org.apache.dolphinscheduler.dao.entity.TaskGroupQueue">
select
<include refid="baseSql" />
from t_ds_task_group_queue
where in_queue = #{inQueue} order by priority desc
</select>
<select id="queryByTaskInstanceId" resultType="org.apache.dolphinscheduler.dao.entity.TaskGroupQueue">
select
<include refid="baseSql" />
from t_ds_task_group_queue
where task_id = #{taskInstanceId}
</select>
<select id="queryUsingTaskGroupQueueByGroupId" resultType="org.apache.dolphinscheduler.dao.entity.TaskGroupQueue">
select
<include refid="baseSql" />
from t_ds_task_group_queue
where group_id = #{taskGroupId} and status = #{status} and force_start = #{forceStart} and in_queue = #{inQueue}
</select>
</mapper>

View File

@ -30,24 +30,6 @@
${alias}.flag, ${alias}.is_cache, ${alias}.cache_key, ${alias}.retry_interval, ${alias}.max_retry_times, ${alias}.task_instance_priority, ${alias}.worker_group,${alias}.environment_code , ${alias}.executor_id,
${alias}.first_submit_time, ${alias}.delay_time, ${alias}.task_params, ${alias}.var_pool, ${alias}.dry_run, ${alias}.test_flag, ${alias}.task_group_id, ${alias}.task_execute_type
</sql>
<update id="setFailoverByHostAndStateArray">
update t_ds_task_instance
set state = #{destStatus}
where host = #{host}
<if test="states != null and states.length != 0">
and state in
<foreach collection="states" index="index" item="i" open="(" separator="," close=")">
#{i}
</foreach>
</if>
</update>
<select id="queryTaskByProcessIdAndState" resultType="java.lang.Integer">
select id
from t_ds_task_instance
WHERE process_instance_id = #{processInstanceId}
and state = #{state}
and flag = 1
</select>
<select id="findValidTaskListByProcessId" resultType="org.apache.dolphinscheduler.dao.entity.TaskInstance">
select
<include refid="baseSql"/>
@ -63,21 +45,6 @@
from t_ds_task_instance
WHERE process_instance_id = #{workflowInstanceId}
</select>
<select id="queryByHostAndStatus" resultType="org.apache.dolphinscheduler.dao.entity.TaskInstance">
select
<include refid="baseSql"/>
from t_ds_task_instance
where 1 = 1
<if test="host != null and host != ''">
and host = #{host}
</if>
<if test="states != null and states.length != 0">
and state in
<foreach collection="states" index="index" item="i" open="(" separator="," close=")">
#{i}
</foreach>
</if>
</select>
<select id="countTaskInstanceStateByProjectCodes" resultType="org.apache.dolphinscheduler.dao.model.TaskInstanceStatusCountDto">
select state, count(0) as count
@ -118,32 +85,7 @@
</if>
group by t.state
</select>
<select id="countTaskInstanceStateByProjectCodesAndStatesBySubmitTime" resultType="org.apache.dolphinscheduler.dao.entity.ExecuteStatusCount">
select
state, count(0) as count
from t_ds_task_instance t
left join t_ds_task_definition_log d on d.code=t.task_code and d.version=t.task_definition_version
where 1=1
<if test="states != null and states.size != 0">
and t.state in
<foreach collection="states" index="index" item="state" open="(" separator="," close=")">
#{state.code}
</foreach>
</if>
<if test="projectCodes != null and projectCodes.length != 0">
and d.project_code in
<foreach collection="projectCodes" index="index" item="i" open="(" separator="," close=")">
#{i}
</foreach>
</if>
<if test="startTime != null">
and t.submit_time <![CDATA[ > ]]> #{startTime}
</if>
<if test="endTime != null">
and t.submit_time <![CDATA[ <= ]]> #{endTime}
</if>
group by t.state
</select>
<select id="countTaskInstanceStateByProjectCodesAndStatesBySubmitTimeV2" resultType="org.apache.dolphinscheduler.dao.entity.ExecuteStatusCount">
select
state, count(0) as count
@ -181,15 +123,7 @@
</if>
group by t.state
</select>
<select id="queryByInstanceIdAndName" resultType="org.apache.dolphinscheduler.dao.entity.TaskInstance">
select
<include refid="baseSql"/>
from t_ds_task_instance
where process_instance_id = #{processInstanceId}
and name = #{name}
and flag = 1
limit 1
</select>
<select id="queryByInstanceIdAndCode" resultType="org.apache.dolphinscheduler.dao.entity.TaskInstance">
select
<include refid="baseSql"/>
@ -229,24 +163,7 @@
</foreach>
</if>
</select>
<select id="countTask" resultType="java.lang.Integer">
select count(1) as count
from t_ds_task_instance task,t_ds_task_definition_log define
where task.task_code=define.code
and task.task_definition_version=define.version
<if test="projectCodes != null and projectCodes.length != 0">
and define.project_code in
<foreach collection="projectCodes" index="index" item="i" open="(" separator="," close=")">
#{i}
</foreach>
</if>
<if test="taskIds != null and taskIds.length != 0">
and task.id in
<foreach collection="taskIds" index="index" item="i" open="(" separator="," close=")">
#{i}
</foreach>
</if>
</select>
<select id="queryTaskInstanceListPaging" resultType="org.apache.dolphinscheduler.dao.entity.TaskInstance">
select
<include refid="baseSql"/>
@ -330,16 +247,6 @@
</if>
order by start_time desc
</select>
<select id="loadAllInfosNoRelease" resultType="org.apache.dolphinscheduler.dao.entity.TaskInstance">
select
<include refid="baseSqlV2">
<property name="alias" value="instance"/>
</include>
from t_ds_task_instance instance
left join t_ds_task_group_queue que on instance.id = que.task_id
where instance.process_instance_id = #{processInstanceId}
and que.status = #{status}
</select>
<select id="findLastTaskInstances" resultType="org.apache.dolphinscheduler.dao.entity.TaskInstance">
select
<include refid="baseSqlV2">

View File

@ -1054,7 +1054,7 @@ CREATE TABLE t_ds_version
-- Records of t_ds_version
-- ----------------------------
INSERT INTO t_ds_version
VALUES ('1', '3.3.0');
VALUES ('1', '3.2.1');
-- ----------------------------
@ -1976,6 +1976,7 @@ CREATE TABLE t_ds_task_group_queue
in_queue int(4) DEFAULT '0' ,
create_time datetime DEFAULT NULL ,
update_time datetime DEFAULT NULL ,
KEY idx_t_ds_task_group_queue_in_queue (in_queue) ,
PRIMARY KEY (id)
);

View File

@ -1045,7 +1045,7 @@ CREATE TABLE `t_ds_version` (
-- ----------------------------
-- Records of t_ds_version
-- ----------------------------
INSERT IGNORE INTO `t_ds_version` VALUES ('1', '3.3.0');
INSERT IGNORE INTO `t_ds_version` VALUES ('1', '3.2.1');
-- ----------------------------
@ -1964,6 +1964,7 @@ CREATE TABLE `t_ds_task_group_queue` (
`in_queue` tinyint(4) DEFAULT '0' COMMENT 'ready to get the queue by other task finish 0 NO ,1 YES',
`create_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP,
`update_time` timestamp NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
KEY `idx_t_ds_task_group_queue_in_queue` (`in_queue`),
PRIMARY KEY( `id` )
)ENGINE= INNODB AUTO_INCREMENT= 1 DEFAULT CHARSET= utf8 COLLATE = utf8_bin;
@ -2091,6 +2092,10 @@ CREATE TABLE `t_ds_fav_task`
AUTO_INCREMENT = 1
DEFAULT CHARSET = utf8 COLLATE = utf8_bin;
-- ----------------------------
-- Table structure for t_ds_trigger_relation
-- ----------------------------
DROP TABLE IF EXISTS `t_ds_trigger_relation`;
CREATE TABLE `t_ds_trigger_relation` (
`id` bigint(20) NOT NULL AUTO_INCREMENT,
`trigger_type` int(11) NOT NULL DEFAULT '0' COMMENT '0 process 1 task',

View File

@ -1043,7 +1043,7 @@ INSERT INTO t_ds_queue(queue_name, queue, create_time, update_time)
VALUES ('default', 'default', '2018-11-29 10:22:33', '2018-11-29 10:22:33');
-- Records of t_ds_queue,default queue name : default
INSERT INTO t_ds_version(version) VALUES ('3.3.0');
INSERT INTO t_ds_version(version) VALUES ('3.2.1');
--
-- Table structure for table t_ds_plugin_define
@ -1946,6 +1946,8 @@ CREATE TABLE t_ds_task_group_queue (
PRIMARY KEY (id)
);
create index idx_t_ds_task_group_queue_in_queue on t_ds_task_group_queue(in_queue);
--
-- Table structure for table t_ds_task_group
--

View File

@ -1 +1 @@
3.3.0
3.2.1

View File

@ -15,6 +15,74 @@
* limitations under the License.
*/
-- Modify "t_ds_alertgroup" table
ALTER TABLE `t_ds_alertgroup` AUTO_INCREMENT 3;
-- Modify "t_ds_alert_plugin_instance" table
ALTER TABLE `t_ds_alert_plugin_instance`
ADD COLUMN `instance_type` int NOT NULL DEFAULT 0, ADD COLUMN `warning_type` int NOT NULL DEFAULT 3;
-- Create "t_ds_listener_event" table
CREATE TABLE `t_ds_listener_event`
(
`id` int NOT NULL AUTO_INCREMENT COMMENT "key",
`content` text NULL COMMENT "listener event json content",
`sign` char(64) NOT NULL DEFAULT "" COMMENT "sign=sha1(content)",
`post_status` tinyint NOT NULL DEFAULT 0 COMMENT "0:wait running,1:success,2:failed,3:partial success",
`event_type` int NOT NULL COMMENT "listener event type",
`log` text NULL COMMENT "log",
`create_time` datetime NULL COMMENT "create time",
`update_time` datetime NULL COMMENT "update time",
PRIMARY KEY (`id`),
INDEX `idx_sign` (`sign`),
INDEX `idx_status` (`post_status`)
) CHARSET utf8 COLLATE utf8_bin;
-- modify_data_t_ds_dq_rule_input_entry behavior change
--DROP PROCEDURE if EXISTS modify_data_t_ds_dq_rule_input_entry;
DROP PROCEDURE if EXISTS modify_data_t_ds_dq_rule_input_entry;
delimiter d//
CREATE PROCEDURE modify_data_t_ds_dq_rule_input_entry()
BEGIN
IF EXISTS (SELECT 1 FROM information_schema.COLUMNS
WHERE TABLE_NAME='t_ds_dq_rule_input_entry'
AND TABLE_SCHEMA=(SELECT DATABASE())
AND COLUMN_NAME ='value')
THEN
ALTER TABLE `t_ds_dq_rule_input_entry`
CHANGE COLUMN `value` `data` varchar(255) DEFAULT NULL;
END IF;
END;
d//
delimiter ;
CALL modify_data_t_ds_dq_rule_input_entry;
DROP PROCEDURE modify_data_t_ds_dq_rule_input_entry;
-- modify_data_value_t_ds_dq_rule_input_entry behavior change
--DROP PROCEDURE if EXISTS modify_data_value_t_ds_dq_rule_input_entry;
DROP PROCEDURE if EXISTS modify_data_value_t_ds_dq_rule_input_entry;
delimiter d//
CREATE PROCEDURE modify_data_value_t_ds_dq_rule_input_entry()
BEGIN
IF EXISTS (SELECT 1 FROM information_schema.COLUMNS
WHERE TABLE_NAME='t_ds_dq_rule_input_entry'
AND TABLE_SCHEMA=(SELECT DATABASE())
AND COLUMN_NAME ='value_type')
THEN
ALTER TABLE `t_ds_dq_rule_input_entry`
CHANGE COLUMN `value_type` `data_type` int(11) DEFAULT NULL;
END IF;
END;
d//
delimiter ;
CALL modify_data_value_t_ds_dq_rule_input_entry;
DROP PROCEDURE modify_data_value_t_ds_dq_rule_input_entry;
ALTER TABLE `t_ds_process_definition` MODIFY COLUMN `version` int NOT NULL DEFAULT 1 COMMENT "process definition version";
ALTER TABLE `t_ds_process_definition_log` MODIFY COLUMN `version` int NOT NULL DEFAULT 1 COMMENT "process definition version";
ALTER TABLE `t_ds_process_instance` MODIFY COLUMN `process_definition_version` int NOT NULL DEFAULT 1 COMMENT "process definition version";
ALTER TABLE `t_ds_task_definition` MODIFY COLUMN `version` int NOT NULL DEFAULT 1 COMMENT "task definition version";
ALTER TABLE `t_ds_task_definition_log` MODIFY COLUMN `version` int NOT NULL DEFAULT 1 COMMENT "task definition version";
ALTER TABLE `t_ds_task_instance` MODIFY COLUMN `task_definition_version` int NOT NULL DEFAULT 1 COMMENT "task definition version";
-- t_ds_k8s_namespace
-- ALTER TABLE t_ds_k8s_namespace DROP COLUMN IF EXISTS limits_cpu;
drop PROCEDURE if EXISTS drop_t_ds_k8s_namespace_col_limits_cpu;
@ -109,4 +177,19 @@ ALTER TABLE `t_ds_process_definition_log` MODIFY COLUMN `version` int NOT NULL D
ALTER TABLE `t_ds_process_instance` MODIFY COLUMN `process_definition_version` int NOT NULL DEFAULT 1 COMMENT "process definition version";
ALTER TABLE `t_ds_task_definition` MODIFY COLUMN `version` int NOT NULL DEFAULT 1 COMMENT "task definition version";
ALTER TABLE `t_ds_task_definition_log` MODIFY COLUMN `version` int NOT NULL DEFAULT 1 COMMENT "task definition version";
ALTER TABLE `t_ds_task_instance` MODIFY COLUMN `task_definition_version` int NOT NULL DEFAULT 1 COMMENT "task definition version";
ALTER TABLE `t_ds_task_instance` MODIFY COLUMN `task_definition_version` int NOT NULL DEFAULT 1 COMMENT "task definition version";
-- create idx_t_ds_task_group_queue_in_queue on t_ds_task_group_queue
DROP PROCEDURE IF EXISTS create_idx_t_ds_task_group_queue_in_queue;
delimiter d//
CREATE PROCEDURE create_idx_t_ds_task_group_queue_in_queue()
BEGIN
DECLARE index_exists INT DEFAULT 0;
SELECT COUNT(*) INTO index_exists FROM information_schema.statistics WHERE table_schema = (SELECT DATABASE()) AND table_name = 't_ds_task_group_queue' AND index_name = 'idx_t_ds_task_group_queue_in_queue';
IF index_exists = 0 THEN CREATE INDEX idx_t_ds_task_group_queue_in_queue ON t_ds_task_group_queue(in_queue);
END IF;
END;
d//
delimiter ;
CALL create_idx_t_ds_task_group_queue_in_queue;
DROP PROCEDURE create_idx_t_ds_task_group_queue_in_queue;

View File

@ -14,3 +14,6 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
INSERT IGNORE INTO `t_ds_alertgroup`(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'global alert group', 'global alert group', current_timestamp, current_timestamp);

View File

@ -14,6 +14,54 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
-- Modify "t_ds_alert_plugin_instance" table
ALTER TABLE "t_ds_alert_plugin_instance" ADD COLUMN "instance_type" integer NOT NULL DEFAULT 0, ADD COLUMN "warning_type" integer NOT NULL DEFAULT 3;
-- Create "t_ds_listener_event" table
CREATE TABLE "t_ds_listener_event" ("id" integer NOT NULL, "content" text NULL, "sign" character varying(64) NOT NULL DEFAULT '', "post_status" integer NOT NULL DEFAULT 0, "event_type" integer NOT NULL, "log" text NULL, "create_time" timestamp NULL, "update_time" timestamp NULL, PRIMARY KEY ("id"));
-- Create index "idx_listener_event_post_status" to table: "t_ds_listener_event"
CREATE INDEX "idx_listener_event_post_status" ON "t_ds_listener_event" ("post_status");
-- Create index "idx_listener_event_sign" to table: "t_ds_listener_event"
CREATE INDEX "idx_listener_event_sign" ON "t_ds_listener_event" ("sign");
-- Set comment to column: "sign" on table: "t_ds_listener_event"
COMMENT ON COLUMN "t_ds_listener_event" ."sign" IS 'sign=sha1(content)';
-- modify_data_t_ds_dq_rule_input_entry
delimiter d//
CREATE OR REPLACE FUNCTION modify_data_t_ds_dq_rule_input_entry() RETURNS void AS $$
BEGIN
IF EXISTS (SELECT 1
FROM information_schema.columns
WHERE table_name = 't_ds_dq_rule_input_entry'
AND column_name = 'value')
THEN
ALTER TABLE t_ds_dq_rule_input_entry
RENAME COLUMN "value" TO "data";
END IF;
END;
$$ LANGUAGE plpgsql;
d//
select modify_data_t_ds_dq_rule_input_entry();
DROP FUNCTION IF EXISTS modify_data_t_ds_dq_rule_input_entry();
-- modify_data_type_t_ds_dq_rule_input_entry
delimiter d//
CREATE OR REPLACE FUNCTION modify_data_type_t_ds_dq_rule_input_entry() RETURNS void AS $$
BEGIN
IF EXISTS (SELECT 1
FROM information_schema.columns
WHERE table_name = 't_ds_dq_rule_input_entry'
AND column_name = 'value_type')
THEN
ALTER TABLE t_ds_dq_rule_input_entry
RENAME COLUMN "value_type" TO "data_type";
END IF;
END;
$$ LANGUAGE plpgsql;
d//
select modify_data_type_t_ds_dq_rule_input_entry();
DROP FUNCTION IF EXISTS modify_data_type_t_ds_dq_rule_input_entry();
-- t_ds_k8s_namespace
ALTER TABLE "t_ds_k8s_namespace" DROP COLUMN IF EXISTS "limits_cpu";
@ -29,4 +77,6 @@ ALTER TABLE "t_ds_process_definition_log" ALTER COLUMN "version" SET DEFAULT 1;
ALTER TABLE "t_ds_task_definition" ALTER COLUMN "version" SET DEFAULT 1;
ALTER TABLE "t_ds_task_definition_log" ALTER COLUMN "version" SET DEFAULT 1;
ALTER TABLE "t_ds_process_instance" ALTER COLUMN "process_definition_version" SET NOT NULL, ALTER COLUMN "process_definition_version" SET DEFAULT 1;
ALTER TABLE "t_ds_task_instance" ALTER COLUMN "task_definition_version" SET NOT NULL, ALTER COLUMN "task_definition_version" SET DEFAULT 1;
ALTER TABLE "t_ds_task_instance" ALTER COLUMN "task_definition_version" SET NOT NULL, ALTER COLUMN "task_definition_version" SET DEFAULT 1;
CREATE INDEX IF NOT EXISTS idx_t_ds_task_group_queue_in_queue ON t_ds_task_group_queue(in_queue);

View File

@ -14,3 +14,5 @@
* See the License for the specific language governing permissions and
* limitations under the License.
*/
INSERT INTO t_ds_alertgroup(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'global alert group', 'global alert group', '2018-11-29 10:20:39', '2018-11-29 10:20:39');

View File

@ -1,83 +0,0 @@
/*
* 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.
*/
-- Modify "t_ds_alertgroup" table
ALTER TABLE `t_ds_alertgroup` AUTO_INCREMENT 3;
-- Modify "t_ds_alert_plugin_instance" table
ALTER TABLE `t_ds_alert_plugin_instance`
ADD COLUMN `instance_type` int NOT NULL DEFAULT 0, ADD COLUMN `warning_type` int NOT NULL DEFAULT 3;
-- Create "t_ds_listener_event" table
CREATE TABLE `t_ds_listener_event`
(
`id` int NOT NULL AUTO_INCREMENT COMMENT "key",
`content` text NULL COMMENT "listener event json content",
`sign` char(64) NOT NULL DEFAULT "" COMMENT "sign=sha1(content)",
`post_status` tinyint NOT NULL DEFAULT 0 COMMENT "0:wait running,1:success,2:failed,3:partial success",
`event_type` int NOT NULL COMMENT "listener event type",
`log` text NULL COMMENT "log",
`create_time` datetime NULL COMMENT "create time",
`update_time` datetime NULL COMMENT "update time",
PRIMARY KEY (`id`),
INDEX `idx_sign` (`sign`),
INDEX `idx_status` (`post_status`)
) CHARSET utf8 COLLATE utf8_bin;
-- modify_data_t_ds_dq_rule_input_entry behavior change
--DROP PROCEDURE if EXISTS modify_data_t_ds_dq_rule_input_entry;
DROP PROCEDURE if EXISTS modify_data_t_ds_dq_rule_input_entry;
delimiter d//
CREATE PROCEDURE modify_data_t_ds_dq_rule_input_entry()
BEGIN
IF EXISTS (SELECT 1 FROM information_schema.COLUMNS
WHERE TABLE_NAME='t_ds_dq_rule_input_entry'
AND TABLE_SCHEMA=(SELECT DATABASE())
AND COLUMN_NAME ='value')
THEN
ALTER TABLE `t_ds_dq_rule_input_entry`
CHANGE COLUMN `value` `data` varchar(255) DEFAULT NULL;
END IF;
END;
d//
delimiter ;
CALL modify_data_t_ds_dq_rule_input_entry;
DROP PROCEDURE modify_data_t_ds_dq_rule_input_entry;
-- modify_data_value_t_ds_dq_rule_input_entry behavior change
--DROP PROCEDURE if EXISTS modify_data_value_t_ds_dq_rule_input_entry;
DROP PROCEDURE if EXISTS modify_data_value_t_ds_dq_rule_input_entry;
delimiter d//
CREATE PROCEDURE modify_data_value_t_ds_dq_rule_input_entry()
BEGIN
IF EXISTS (SELECT 1 FROM information_schema.COLUMNS
WHERE TABLE_NAME='t_ds_dq_rule_input_entry'
AND TABLE_SCHEMA=(SELECT DATABASE())
AND COLUMN_NAME ='value_type')
THEN
ALTER TABLE `t_ds_dq_rule_input_entry`
CHANGE COLUMN `value_type` `data_type` int(11) DEFAULT NULL;
END IF;
END;
d//
delimiter ;
CALL modify_data_value_t_ds_dq_rule_input_entry;
DROP PROCEDURE modify_data_value_t_ds_dq_rule_input_entry;
ALTER TABLE `t_ds_process_definition` MODIFY COLUMN `version` int NOT NULL DEFAULT 1 COMMENT "process definition version";
ALTER TABLE `t_ds_process_definition_log` MODIFY COLUMN `version` int NOT NULL DEFAULT 1 COMMENT "process definition version";
ALTER TABLE `t_ds_process_instance` MODIFY COLUMN `process_definition_version` int NOT NULL DEFAULT 1 COMMENT "process definition version";
ALTER TABLE `t_ds_task_definition` MODIFY COLUMN `version` int NOT NULL DEFAULT 1 COMMENT "task definition version";
ALTER TABLE `t_ds_task_definition_log` MODIFY COLUMN `version` int NOT NULL DEFAULT 1 COMMENT "task definition version";
ALTER TABLE `t_ds_task_instance` MODIFY COLUMN `task_definition_version` int NOT NULL DEFAULT 1 COMMENT "task definition version";

View File

@ -1,16 +0,0 @@
/*
* 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.
*/

View File

@ -1,18 +0,0 @@
/*
* 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.
*/
INSERT IGNORE INTO `t_ds_alertgroup`(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'global alert group', 'global alert group', current_timestamp, current_timestamp);

View File

@ -1,71 +0,0 @@
/*
* 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.
*/
-- Modify "t_ds_alert_plugin_instance" table
ALTER TABLE "t_ds_alert_plugin_instance" ADD COLUMN "instance_type" integer NOT NULL DEFAULT 0, ADD COLUMN "warning_type" integer NOT NULL DEFAULT 3;
-- Create "t_ds_listener_event" table
CREATE TABLE "t_ds_listener_event" ("id" integer NOT NULL, "content" text NULL, "sign" character varying(64) NOT NULL DEFAULT '', "post_status" integer NOT NULL DEFAULT 0, "event_type" integer NOT NULL, "log" text NULL, "create_time" timestamp NULL, "update_time" timestamp NULL, PRIMARY KEY ("id"));
-- Create index "idx_listener_event_post_status" to table: "t_ds_listener_event"
CREATE INDEX "idx_listener_event_post_status" ON "t_ds_listener_event" ("post_status");
-- Create index "idx_listener_event_sign" to table: "t_ds_listener_event"
CREATE INDEX "idx_listener_event_sign" ON "t_ds_listener_event" ("sign");
-- Set comment to column: "sign" on table: "t_ds_listener_event"
COMMENT ON COLUMN "t_ds_listener_event" ."sign" IS 'sign=sha1(content)';
-- modify_data_t_ds_dq_rule_input_entry
delimiter d//
CREATE OR REPLACE FUNCTION modify_data_t_ds_dq_rule_input_entry() RETURNS void AS $$
BEGIN
IF EXISTS (SELECT 1
FROM information_schema.columns
WHERE table_name = 't_ds_dq_rule_input_entry'
AND column_name = 'value')
THEN
ALTER TABLE t_ds_dq_rule_input_entry
RENAME COLUMN "value" TO "data";
END IF;
END;
$$ LANGUAGE plpgsql;
d//
select modify_data_t_ds_dq_rule_input_entry();
DROP FUNCTION IF EXISTS modify_data_t_ds_dq_rule_input_entry();
-- modify_data_type_t_ds_dq_rule_input_entry
delimiter d//
CREATE OR REPLACE FUNCTION modify_data_type_t_ds_dq_rule_input_entry() RETURNS void AS $$
BEGIN
IF EXISTS (SELECT 1
FROM information_schema.columns
WHERE table_name = 't_ds_dq_rule_input_entry'
AND column_name = 'value_type')
THEN
ALTER TABLE t_ds_dq_rule_input_entry
RENAME COLUMN "value_type" TO "data_type";
END IF;
END;
$$ LANGUAGE plpgsql;
d//
select modify_data_type_t_ds_dq_rule_input_entry();
DROP FUNCTION IF EXISTS modify_data_type_t_ds_dq_rule_input_entry();
ALTER TABLE "t_ds_process_definition" ALTER COLUMN "version" SET DEFAULT 1;
ALTER TABLE "t_ds_process_definition_log" ALTER COLUMN "version" SET DEFAULT 1;
ALTER TABLE "t_ds_task_definition" ALTER COLUMN "version" SET DEFAULT 1;
ALTER TABLE "t_ds_task_definition_log" ALTER COLUMN "version" SET DEFAULT 1;
ALTER TABLE "t_ds_process_instance" ALTER COLUMN "process_definition_version" SET NOT NULL, ALTER COLUMN "process_definition_version" SET DEFAULT 1;
ALTER TABLE "t_ds_task_instance" ALTER COLUMN "task_definition_version" SET NOT NULL, ALTER COLUMN "task_definition_version" SET DEFAULT 1;

View File

@ -1,16 +0,0 @@
/*
* 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.
*/

View File

@ -1,18 +0,0 @@
/*
* 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.
*/
INSERT INTO t_ds_alertgroup(alert_instance_ids, create_user_id, group_name, description, create_time, update_time)
VALUES (NULL, 1, 'global alert group', 'global alert group', '2018-11-29 10:20:39', '2018-11-29 10:20:39');

View File

@ -197,30 +197,6 @@ public class ProcessInstanceMapperTest extends BaseDaoTest {
processInstanceMapper.deleteById(processInstance.getId());
}
/**
* test set failover by host and state
*/
@Test
public void testSetFailoverByHostAndStateArray() {
int[] stateArray = new int[]{
WorkflowExecutionStatus.RUNNING_EXECUTION.ordinal(),
WorkflowExecutionStatus.SUCCESS.ordinal()};
ProcessInstance processInstance = insertOne();
processInstance.setState(WorkflowExecutionStatus.RUNNING_EXECUTION);
processInstance.setHost("192.168.2.220");
processInstanceMapper.updateById(processInstance);
String host = processInstance.getHost();
int update = processInstanceMapper.setFailoverByHostAndStateArray(host, stateArray);
Assertions.assertNotEquals(0, update);
processInstance = processInstanceMapper.selectById(processInstance.getId());
Assertions.assertNull(processInstance.getHost());
processInstanceMapper.deleteById(processInstance.getId());
}
/**
* test update process instance by state
*/

View File

@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.dao.mapper;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.dao.BaseDaoTest;
import org.apache.dolphinscheduler.dao.entity.TaskGroup;
@ -24,8 +25,6 @@ import java.util.Date;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import com.baomidou.mybatisplus.core.metadata.IPage;
@ -33,8 +32,6 @@ import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
public class TaskGroupMapperTest extends BaseDaoTest {
private static final Logger logger = LoggerFactory.getLogger(TaskGroupMapperTest.class);
@Autowired
TaskGroupMapper taskGroupMapper;
@ -46,7 +43,7 @@ public class TaskGroupMapperTest extends BaseDaoTest {
taskGroup.setName("task group");
taskGroup.setId(1);
taskGroup.setUserId(1);
taskGroup.setStatus(1);
taskGroup.setStatus(Flag.YES);
taskGroup.setGroupSize(10);
taskGroup.setDescription("this is a task group");
Date date = new Date(System.currentTimeMillis());
@ -88,7 +85,7 @@ public class TaskGroupMapperTest extends BaseDaoTest {
Page<TaskGroup> page = new Page(1, 3);
IPage<TaskGroup> taskGroupIPage = taskGroupMapper.queryTaskGroupPaging(
page,
taskGroup.getName(), taskGroup.getStatus());
taskGroup.getName(), taskGroup.getStatus().getCode());
Assertions.assertEquals(taskGroupIPage.getTotal(), 1);
}

View File

@ -143,25 +143,6 @@ public class TaskInstanceMapperTest extends BaseDaoTest {
Assertions.assertNotEquals(0, taskInstances.size());
}
/**
* test query task instance by process instance id and state
*/
@Test
public void testQueryTaskByProcessIdAndState() {
// insert ProcessInstance
ProcessInstance processInstance = insertProcessInstance();
// insert taskInstance
TaskInstance task = insertTaskInstance(processInstance.getId());
task.setProcessInstanceId(processInstance.getId());
taskInstanceMapper.updateById(task);
List<Integer> taskInstances = taskInstanceMapper.queryTaskByProcessIdAndState(
task.getProcessInstanceId(),
TaskExecutionStatus.RUNNING_EXECUTION.getCode());
taskInstanceMapper.deleteById(task.getId());
Assertions.assertNotEquals(0, taskInstances.size());
}
/**
* test find valid task list by process instance id
*/
@ -194,66 +175,6 @@ public class TaskInstanceMapperTest extends BaseDaoTest {
Assertions.assertNotEquals(0, taskInstances1.size());
}
/**
* test query by host and status
*/
@Test
public void testQueryByHostAndStatus() {
// insert ProcessInstance
ProcessInstance processInstance = insertProcessInstance();
// insert taskInstance
TaskInstance task = insertTaskInstance(processInstance.getId());
task.setHost("111.111.11.11");
taskInstanceMapper.updateById(task);
List<TaskInstance> taskInstances = taskInstanceMapper.queryByHostAndStatus(
task.getHost(), new int[]{TaskExecutionStatus.RUNNING_EXECUTION.getCode()});
taskInstanceMapper.deleteById(task.getId());
Assertions.assertNotEquals(0, taskInstances.size());
}
/**
* test set failover by host and state array
*/
@Test
public void testSetFailoverByHostAndStateArray() {
// insert ProcessInstance
ProcessInstance processInstance = insertProcessInstance();
// insert taskInstance
TaskInstance task = insertTaskInstance(processInstance.getId());
task.setHost("111.111.11.11");
taskInstanceMapper.updateById(task);
int setResult = taskInstanceMapper.setFailoverByHostAndStateArray(
task.getHost(),
new int[]{TaskExecutionStatus.RUNNING_EXECUTION.getCode()},
TaskExecutionStatus.NEED_FAULT_TOLERANCE);
taskInstanceMapper.deleteById(task.getId());
Assertions.assertNotEquals(0, setResult);
}
/**
* test query by task instance id and name
*/
@Test
public void testQueryByInstanceIdAndName() {
// insert ProcessInstance
ProcessInstance processInstance = insertProcessInstance();
// insert taskInstance
TaskInstance task = insertTaskInstance(processInstance.getId());
task.setHost("111.111.11.11");
taskInstanceMapper.updateById(task);
TaskInstance taskInstance = taskInstanceMapper.queryByInstanceIdAndName(
task.getProcessInstanceId(),
task.getName());
taskInstanceMapper.deleteById(task.getId());
Assertions.assertNotEquals(null, taskInstance);
}
/**
* test query by task instance id and code
*/
@ -294,37 +215,6 @@ public class TaskInstanceMapperTest extends BaseDaoTest {
Assertions.assertEquals(1, taskInstances.size());
}
/**
* test count task instance
*/
@Test
public void testCountTask() {
// insert ProcessInstance
ProcessInstance processInstance = insertProcessInstance();
// insert taskInstance
TaskInstance task = insertTaskInstance(processInstance.getId());
ProcessDefinition definition = new ProcessDefinition();
definition.setCode(1L);
definition.setProjectCode(1111L);
definition.setCreateTime(new Date());
definition.setUpdateTime(new Date());
processDefinitionMapper.insert(definition);
taskInstanceMapper.updateById(task);
int countTask = taskInstanceMapper.countTask(
new Long[0],
new int[0]);
int countTask2 = taskInstanceMapper.countTask(
new Long[]{definition.getProjectCode()},
new int[]{task.getId()});
taskInstanceMapper.deleteById(task.getId());
processDefinitionMapper.deleteById(definition.getId());
Assertions.assertEquals(0, countTask);
Assertions.assertEquals(0, countTask2);
}
/**
* test count task instance state by user
*/

View File

@ -0,0 +1,118 @@
/*
* 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 static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.dao.BaseDaoTest;
import org.apache.dolphinscheduler.dao.entity.TaskGroup;
import org.apache.dolphinscheduler.dao.repository.TaskGroupDao;
import java.util.Date;
import java.util.List;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
class TaskGroupDaoImplTest extends BaseDaoTest {
@Autowired
private TaskGroupDao taskGroupDao;
@Test
void queryAllTaskGroups() {
TaskGroup taskGroup = createTaskGroup("test", 0, 1);
taskGroupDao.insert(taskGroup);
List<TaskGroup> taskGroups = taskGroupDao.queryAllTaskGroups();
assertEquals(1, taskGroups.size());
}
@Test
void queryUsedTaskGroups() {
// Insert a unused task group
TaskGroup taskGroup = createTaskGroup("testUnused", 0, 1);
taskGroupDao.insert(taskGroup);
assertEquals(0, taskGroupDao.queryUsedTaskGroups().size());
// Insert a used task group
taskGroup = createTaskGroup("testUsed", 1, 1);
taskGroupDao.insert(taskGroup);
assertEquals(1, taskGroupDao.queryUsedTaskGroups().size());
}
@Test
void queryAvailableTaskGroups() {
// Insert a full task group
TaskGroup taskGroup = createTaskGroup("testFull", 1, 1);
taskGroupDao.insert(taskGroup);
assertEquals(0, taskGroupDao.queryAvailableTaskGroups().size());
// Insert a used task group
taskGroup = createTaskGroup("testNotFull", 0, 1);
taskGroupDao.insert(taskGroup);
assertEquals(1, taskGroupDao.queryAvailableTaskGroups().size());
}
@Test
void acquireTaskGroupSlot() {
// Insert a full task group will acquire failed
TaskGroup taskGroup = createTaskGroup("testFull", 1, 1);
taskGroupDao.insert(taskGroup);
assertFalse(taskGroupDao.acquireTaskGroupSlot(taskGroup.getId()));
taskGroup.setUseSize(0);
taskGroupDao.updateById(taskGroup);
assertTrue(taskGroupDao.acquireTaskGroupSlot(taskGroup.getId()));
taskGroup = taskGroupDao.queryById(taskGroup.getId());
assertEquals(1, taskGroup.getUseSize());
}
@Test
void releaseTaskGroupSlot() {
// Insert an empty task group will release failed
TaskGroup taskGroup = createTaskGroup("testEmpty", 0, 1);
taskGroupDao.insert(taskGroup);
assertFalse(taskGroupDao.releaseTaskGroupSlot(taskGroup.getId()));
taskGroup.setUseSize(1);
taskGroupDao.updateById(taskGroup);
assertTrue(taskGroupDao.releaseTaskGroupSlot(taskGroup.getId()));
taskGroup = taskGroupDao.queryById(taskGroup.getId());
assertEquals(0, taskGroup.getUseSize());
}
private TaskGroup createTaskGroup(String name, int useSize, int groupSize) {
return TaskGroup.builder()
.name(name)
.description("test")
.groupSize(groupSize)
.useSize(useSize)
.userId(1)
.status(Flag.YES)
.createTime(new Date())
.updateTime(new Date())
.projectCode(1)
.build();
}
}

View File

@ -0,0 +1,108 @@
/*
* 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 static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import org.apache.dolphinscheduler.common.enums.Flag;
import org.apache.dolphinscheduler.common.enums.TaskGroupQueueStatus;
import org.apache.dolphinscheduler.dao.BaseDaoTest;
import org.apache.dolphinscheduler.dao.entity.TaskGroupQueue;
import org.apache.dolphinscheduler.dao.repository.TaskGroupQueueDao;
import java.util.Date;
import org.assertj.core.util.Lists;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
class TaskGroupQueueDaoImplTest extends BaseDaoTest {
@Autowired
private TaskGroupQueueDao taskGroupQueueDao;
@Test
void deleteByWorkflowInstanceIds() {
TaskGroupQueue taskGroupQueue = createTaskGroupQueue(Flag.NO, TaskGroupQueueStatus.ACQUIRE_SUCCESS);
taskGroupQueueDao.insert(taskGroupQueue);
assertNotNull(taskGroupQueueDao.queryById(taskGroupQueue.getId()));
taskGroupQueueDao.deleteByWorkflowInstanceIds(Lists.newArrayList(1));
assertNull(taskGroupQueueDao.queryById(taskGroupQueue.getId()));
}
@Test
void queryAllInQueueTaskGroupQueue() {
TaskGroupQueue taskGroupQueue = createTaskGroupQueue(Flag.NO, TaskGroupQueueStatus.ACQUIRE_SUCCESS);
taskGroupQueueDao.insert(taskGroupQueue);
assertEquals(1, taskGroupQueueDao.queryAllInQueueTaskGroupQueue().size());
}
@Test
void queryAllInQueueTaskGroupQueueByGroupId() {
TaskGroupQueue taskGroupQueue = createTaskGroupQueue(Flag.NO, TaskGroupQueueStatus.ACQUIRE_SUCCESS);
taskGroupQueueDao.insert(taskGroupQueue);
assertEquals(1, taskGroupQueueDao.queryAllInQueueTaskGroupQueueByGroupId(1).size());
}
@Test
void updateById() {
TaskGroupQueue taskGroupQueue = createTaskGroupQueue(Flag.NO, TaskGroupQueueStatus.WAIT_QUEUE);
taskGroupQueueDao.insert(taskGroupQueue);
taskGroupQueue.setStatus(TaskGroupQueueStatus.ACQUIRE_SUCCESS);
taskGroupQueueDao.updateById(taskGroupQueue);
assertEquals(TaskGroupQueueStatus.ACQUIRE_SUCCESS,
taskGroupQueueDao.queryById(taskGroupQueue.getId()).getStatus());
}
@Test
void queryByTaskInstanceId() {
TaskGroupQueue taskGroupQueue = createTaskGroupQueue(Flag.NO, TaskGroupQueueStatus.ACQUIRE_SUCCESS);
taskGroupQueueDao.insert(taskGroupQueue);
assertEquals(1, taskGroupQueueDao.queryByTaskInstanceId(1).size());
}
@Test
void queryUsingTaskGroupQueueByGroupId() {
TaskGroupQueue taskGroupQueue = createTaskGroupQueue(Flag.NO, TaskGroupQueueStatus.ACQUIRE_SUCCESS);
taskGroupQueueDao.insert(taskGroupQueue);
assertEquals(1, taskGroupQueueDao.queryAcquiredTaskGroupQueueByGroupId(1).size());
taskGroupQueue = createTaskGroupQueue(Flag.YES, TaskGroupQueueStatus.WAIT_QUEUE);
taskGroupQueueDao.insert(taskGroupQueue);
assertEquals(1, taskGroupQueueDao.queryAcquiredTaskGroupQueueByGroupId(1).size());
}
private TaskGroupQueue createTaskGroupQueue(Flag forceStart, TaskGroupQueueStatus taskGroupQueueStatus) {
return TaskGroupQueue.builder()
.taskId(1)
.taskName("test")
.groupId(1)
.processId(1)
.priority(0)
.forceStart(forceStart.getCode())
.inQueue(Flag.YES.getCode())
.status(taskGroupQueueStatus)
.createTime(new Date())
.updateTime(new Date())
.build();
}
}

View File

@ -18,7 +18,7 @@
package org.apache.dolphinscheduler.plugin.datasource.api.utils;
import static org.apache.dolphinscheduler.common.constants.Constants.RESOURCE_STORAGE_TYPE;
import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.DATA_QUALITY_JAR_NAME;
import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.DATA_QUALITY_JAR_DIR;
import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.HADOOP_SECURITY_AUTHENTICATION;
import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.HADOOP_SECURITY_AUTHENTICATION_STARTUP_STATE;
import static org.apache.dolphinscheduler.plugin.task.api.TaskConstants.JAVA_SECURITY_KRB5_CONF;
@ -36,17 +36,26 @@ import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.security.UserGroupInformation;
import java.io.File;
import java.io.IOException;
import java.util.Optional;
import lombok.extern.slf4j.Slf4j;
import org.springframework.core.io.ClassPathResource;
/**
* common utils
*/
@Slf4j
public class CommonUtils {
private CommonUtils() {
throw new UnsupportedOperationException("Construct CommonUtils");
}
private static String DEFAULT_DATA_QUALITY_JAR_PATH = null;
private static final boolean IS_DEVELOP_MODE = PropertyUtils.getBoolean(Constants.DEVELOPMENT_STATE, true);
/**
@ -123,14 +132,71 @@ public class CommonUtils {
return false;
}
public static String getDataQualityJarName() {
String dqsJarName = PropertyUtils.getString(DATA_QUALITY_JAR_NAME);
public static String getDataQualityJarPath() {
log.info("Trying to get data quality jar in path");
String dqJarDir = PropertyUtils.getString(DATA_QUALITY_JAR_DIR);
if (StringUtils.isEmpty(dqsJarName)) {
return "dolphinscheduler-data-quality.jar";
if (StringUtils.isNotEmpty(dqJarDir)) {
log.info(
"Configuration data-quality.jar.dir is not empty, will try to get data quality jar from directory {}",
dqJarDir);
getDataQualityJarPathFromPath(dqJarDir).ifPresent(jarName -> DEFAULT_DATA_QUALITY_JAR_PATH = jarName);
}
return dqsJarName;
if (StringUtils.isEmpty(DEFAULT_DATA_QUALITY_JAR_PATH)) {
log.info("data quality jar path is empty, will try to auto discover it from build-in rules.");
getDefaultDataQualityJarPath();
}
if (StringUtils.isEmpty(DEFAULT_DATA_QUALITY_JAR_PATH)) {
log.error(
"Can not find data quality jar in both configuration and auto discover, please check your configuration or report a bug.");
throw new RuntimeException("data quality jar path is empty");
}
return DEFAULT_DATA_QUALITY_JAR_PATH;
}
private static String getDefaultDataQualityJarPath() {
if (StringUtils.isNotEmpty(DEFAULT_DATA_QUALITY_JAR_PATH)) {
return DEFAULT_DATA_QUALITY_JAR_PATH;
}
try {
// not standalone mode
String currentAbsolutePath = new ClassPathResource("./").getFile().getAbsolutePath();
String currentLibPath = currentAbsolutePath + "/../libs";
getDataQualityJarPathFromPath(currentLibPath).ifPresent(jarName -> DEFAULT_DATA_QUALITY_JAR_PATH = jarName);
// standalone mode
if (StringUtils.isEmpty(DEFAULT_DATA_QUALITY_JAR_PATH)) {
log.info(
"Can not get data quality jar from path {}, maybe service running in standalone mode, will try to find another path",
currentLibPath);
currentLibPath = currentAbsolutePath + "/../../worker-server/libs";
getDataQualityJarPathFromPath(currentLibPath)
.ifPresent(jarName -> DEFAULT_DATA_QUALITY_JAR_PATH = jarName);
}
} catch (IOException e) {
throw new RuntimeException("get default data quality jar path error", e);
}
log.info("get default data quality jar name: {}", DEFAULT_DATA_QUALITY_JAR_PATH);
return DEFAULT_DATA_QUALITY_JAR_PATH;
}
private static Optional<String> getDataQualityJarPathFromPath(String path) {
log.info("Try to get data quality jar from path {}", path);
File[] jars = new File(path).listFiles();
if (jars == null) {
log.warn("No any files find given path {}", path);
return Optional.empty();
}
for (File jar : jars) {
if (jar.getName().startsWith("dolphinscheduler-data-quality")) {
return Optional.of(jar.getAbsolutePath());
}
}
log.warn("No data quality related jar found from path {}", path);
return Optional.empty();
}
/**

View File

@ -33,6 +33,7 @@ import org.apache.commons.collections4.MapUtils;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Properties;
@ -115,6 +116,10 @@ public class MySQLDataSourceProcessor extends AbstractDataSourceProcessor {
@Override
public String getJdbcUrl(ConnectionParam connectionParam) {
MySQLConnectionParam mysqlConnectionParam = (MySQLConnectionParam) connectionParam;
if (MapUtils.isNotEmpty(mysqlConnectionParam.getOther())) {
return String.format("%s?%s", mysqlConnectionParam.getJdbcUrl(),
transformOther(mysqlConnectionParam.getOther()));
}
return mysqlConnectionParam.getJdbcUrl();
}
@ -182,4 +187,13 @@ public class MySQLDataSourceProcessor extends AbstractDataSourceProcessor {
&& !key.contains(ALLOW_URL_IN_LOCAL_IN_FILE_NAME);
}
private String transformOther(Map<String, String> otherMap) {
if (MapUtils.isNotEmpty(otherMap)) {
List<String> list = new ArrayList<>(otherMap.size());
otherMap.forEach((key, value) -> list.add(String.format("%s=%s", key, value)));
return String.join("&", list);
}
return null;
}
}

View File

@ -67,6 +67,23 @@
<fileSet>
<directory>${basedir}/../dolphinscheduler-tools/target/tools</directory>
<outputDirectory>tools</outputDirectory>
<excludes>
<exclude>libs/</exclude>
<exclude>bin/</exclude>
<exclude>dist-bin/</exclude>
</excludes>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-tools/target/tools/libs</directory>
<outputDirectory>tools/libs</outputDirectory>
<includes>
<include>dolphinscheduler-*.jar</include>
<include>spring-*.jar</include>
</includes>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-tools/target/tools/dist-bin</directory>
<outputDirectory>tools/bin</outputDirectory>
</fileSet>
<fileSet>
@ -74,11 +91,6 @@
<outputDirectory>.</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-ui/dist</directory>
<outputDirectory>./ui</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../script</directory>
<outputDirectory>bin</outputDirectory>

View File

@ -95,8 +95,10 @@ datasource.encryption.enable=false
# datasource encryption salt
datasource.encryption.salt=!@#$%^&*
# data quality option
data-quality.jar.name=dolphinscheduler-data-quality-dev-SNAPSHOT.jar
# data quality jar directory path, it would auto discovery data quality jar from this given dir. You should keep it empty if you do not change anything in
# data-quality, it will auto discovery by dolphinscheduler itself. Change it only if you want to use your own data-quality jar and it is not in worker-server
# libs directory(but may sure your jar name start with `dolphinscheduler-data-quality`).
data-quality.jar.dir=
#data-quality.error.output.path=/tmp/data-quality-error-data

View File

@ -25,9 +25,6 @@ import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskKillReque
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskKillResponse;
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskPauseRequest;
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskPauseResponse;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceForceStartRequest;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceForceStartResponse;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceWakeupRequest;
@RpcService
public interface ILogicTaskInstanceOperator {
@ -41,10 +38,4 @@ public interface ILogicTaskInstanceOperator {
@RpcMethod
LogicTaskPauseResponse pauseLogicTask(LogicTaskPauseRequest taskPauseRequest);
@RpcMethod
TaskInstanceForceStartResponse forceStartTaskInstance(TaskInstanceForceStartRequest taskForceStartRequest);
@RpcMethod
void wakeupTaskInstance(TaskInstanceWakeupRequest taskWakeupRequest);
}

View File

@ -20,6 +20,8 @@ package org.apache.dolphinscheduler.extract.master;
import org.apache.dolphinscheduler.extract.base.RpcMethod;
import org.apache.dolphinscheduler.extract.base.RpcService;
import org.apache.dolphinscheduler.extract.master.dto.WorkflowExecuteDto;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceWakeupRequest;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceWakeupResponse;
@RpcService
public interface IWorkflowInstanceService {
@ -30,4 +32,7 @@ public interface IWorkflowInstanceService {
@RpcMethod
WorkflowExecuteDto getWorkflowExecutingData(Integer workflowInstanceId);
@RpcMethod
TaskInstanceWakeupResponse wakeupTaskInstance(TaskInstanceWakeupRequest taskWakeupRequest);
}

View File

@ -1,41 +0,0 @@
/*
* 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.extract.master.transportor;
import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@NoArgsConstructor
@AllArgsConstructor
public class TaskInstanceForceStartResponse {
private boolean success;
private String message;
public static TaskInstanceForceStartResponse success() {
return new TaskInstanceForceStartResponse(true, "dispatch success");
}
public static TaskInstanceForceStartResponse failed(String message) {
return new TaskInstanceForceStartResponse(false, message);
}
}

View File

@ -17,26 +17,21 @@
package org.apache.dolphinscheduler.extract.master.transportor;
import java.io.Serializable;
import lombok.AllArgsConstructor;
import lombok.Builder;
import lombok.Data;
import lombok.NoArgsConstructor;
@Data
@Builder
@NoArgsConstructor
public class TaskInstanceWakeupRequest {
private String key;
@AllArgsConstructor
public class TaskInstanceWakeupRequest implements Serializable {
private int processInstanceId;
private int taskInstanceId;
public TaskInstanceWakeupRequest(
int processInstanceId,
int taskInstanceId) {
this.key = String.format("%d-%d", processInstanceId, taskInstanceId);
this.processInstanceId = processInstanceId;
this.taskInstanceId = taskInstanceId;
}
}

View File

@ -66,12 +66,7 @@
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-task-all</artifactId>
</dependency>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-storage-all</artifactId>
<artifactId>dolphinscheduler-task-all-prune</artifactId>
</dependency>
<dependency>

View File

@ -24,6 +24,7 @@
-XX:+PrintGCDetails
-Xloggc:gc.log
-XX:+ExitOnOutOfMemoryError
-XX:+HeapDumpOnOutOfMemoryError
-XX:HeapDumpPath=dump.hprof

View File

@ -33,6 +33,7 @@ import org.apache.dolphinscheduler.server.master.rpc.MasterRpcServer;
import org.apache.dolphinscheduler.server.master.runner.EventExecuteService;
import org.apache.dolphinscheduler.server.master.runner.FailoverExecuteThread;
import org.apache.dolphinscheduler.server.master.runner.MasterSchedulerBootstrap;
import org.apache.dolphinscheduler.server.master.runner.taskgroup.TaskGroupCoordinator;
import org.apache.dolphinscheduler.service.bean.SpringApplicationContext;
import javax.annotation.PostConstruct;
@ -84,6 +85,9 @@ public class MasterServer implements IStoppable {
@Autowired
private MasterSlotManager masterSlotManager;
@Autowired
private TaskGroupCoordinator taskGroupCoordinator;
public static void main(String[] args) {
MasterServerMetrics.registerUncachedException(DefaultUncaughtExceptionHandler::getUncaughtExceptionCount);
@ -115,6 +119,7 @@ public class MasterServer implements IStoppable {
this.failoverExecuteThread.start();
this.schedulerApi.start();
this.taskGroupCoordinator.start();
MasterServerMetrics.registerMasterCpuUsageGauge(() -> {
SystemMetrics systemMetrics = metricsProvider.getSystemMetrics();

View File

@ -24,14 +24,20 @@ import java.util.Map;
import java.util.Optional;
import java.util.ServiceLoader;
import lombok.extern.slf4j.Slf4j;
@Slf4j
public class StateEventHandlerManager {
private static final Map<StateEventType, StateEventHandler> stateEventHandlerMap = new HashMap<>();
static {
ServiceLoader.load(StateEventHandler.class)
.forEach(stateEventHandler -> stateEventHandlerMap.put(stateEventHandler.getEventType(),
stateEventHandler));
.forEach(stateEventHandler -> {
log.info("Initialize StateEventHandler: {} for eventType: {}",
stateEventHandler.getClass().getName(), stateEventHandler.getEventType());
stateEventHandlerMap.put(stateEventHandler.getEventType(), stateEventHandler);
});
}
public static Optional<StateEventHandler> getStateEventHandler(StateEventType stateEventType) {

View File

@ -62,7 +62,8 @@ public class TaskStateEventHandler implements StateEventHandler {
if (task.getState().isFinished()
&& (taskStateEvent.getStatus() != null && taskStateEvent.getStatus().isRunning())) {
String errorMessage = String.format(
"The current task instance state is %s, but the task state event status is %s, so the task state event will be ignored",
"The current TaskInstance: %s state is %s, but the task state event status is %s, so the task state event will be ignored",
task.getName(),
task.getState().name(),
taskStateEvent.getStatus().name());
log.warn(errorMessage);
@ -75,14 +76,6 @@ public class TaskStateEventHandler implements StateEventHandler {
return true;
}
workflowExecuteRunnable.taskFinished(task);
if (task.getTaskGroupId() > 0) {
log.info("The task instance need to release task Group: {}", task.getTaskGroupId());
try {
workflowExecuteRunnable.releaseTaskGroup(task);
} catch (Exception e) {
throw new StateEventHandleException("Release task group failed", e);
}
}
return true;
}
return true;

View File

@ -1,47 +0,0 @@
/*
* 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.event;
import org.apache.dolphinscheduler.common.enums.StateEventType;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable;
import lombok.extern.slf4j.Slf4j;
import com.google.auto.service.AutoService;
@AutoService(StateEventHandler.class)
@Slf4j
public class TaskWaitTaskGroupStateHandler implements StateEventHandler {
@Override
public boolean handleStateEvent(WorkflowExecuteRunnable workflowExecuteRunnable,
StateEvent stateEvent) {
log.info("Handle task instance wait task group event, taskInstanceId: {}", stateEvent.getTaskInstanceId());
if (workflowExecuteRunnable.checkForceStartAndWakeUp(stateEvent)) {
log.info("Success wake up task instance, taskInstanceId: {}", stateEvent.getTaskInstanceId());
} else {
log.info("Failed to wake up task instance, taskInstanceId: {}", stateEvent.getTaskInstanceId());
}
return true;
}
@Override
public StateEventType getEventType() {
return StateEventType.WAKE_UP_TASK_GROUP;
}
}

View File

@ -24,7 +24,7 @@ import java.util.function.Supplier;
import lombok.experimental.UtilityClass;
import com.facebook.presto.jdbc.internal.guava.collect.ImmutableSet;
import com.google.common.collect.ImmutableSet;
import io.micrometer.core.instrument.Counter;
import io.micrometer.core.instrument.Gauge;

View File

@ -32,12 +32,6 @@ public class LogicTaskInstanceOperationFunctionManager {
@Autowired
private LogicITaskInstancePauseOperationFunction logicITaskInstancePauseOperationFunction;
@Autowired
private TaskInstanceForceStartOperationFunction taskInstanceForceStartOperationFunction;
@Autowired
private TaskInstanceWakeupOperationFunction taskInstanceWakeupOperationFunction;
public LogicITaskInstanceDispatchOperationFunction getLogicTaskInstanceDispatchOperationFunction() {
return logicITaskInstanceDispatchOperationFunction;
}
@ -50,11 +44,4 @@ public class LogicTaskInstanceOperationFunctionManager {
return logicITaskInstancePauseOperationFunction;
}
public TaskInstanceForceStartOperationFunction getTaskInstanceForceStartOperationFunction() {
return taskInstanceForceStartOperationFunction;
}
public TaskInstanceWakeupOperationFunction getTaskInstanceWakeupOperationFunction() {
return taskInstanceWakeupOperationFunction;
}
}

View File

@ -24,9 +24,6 @@ import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskKillReque
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskKillResponse;
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskPauseRequest;
import org.apache.dolphinscheduler.extract.master.transportor.LogicTaskPauseResponse;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceForceStartRequest;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceForceStartResponse;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceWakeupRequest;
import lombok.extern.slf4j.Slf4j;
@ -58,15 +55,4 @@ public class LogicTaskInstanceOperatorImpl implements ILogicTaskInstanceOperator
.operate(taskPauseRequest);
}
@Override
public TaskInstanceForceStartResponse forceStartTaskInstance(TaskInstanceForceStartRequest taskForceStartRequest) {
return logicTaskInstanceOperationFunctionManager.getTaskInstanceForceStartOperationFunction()
.operate(taskForceStartRequest);
}
@Override
public void wakeupTaskInstance(TaskInstanceWakeupRequest taskWakeupRequest) {
logicTaskInstanceOperationFunctionManager.getTaskInstanceWakeupOperationFunction().operate(taskWakeupRequest);
}
}

View File

@ -1,58 +0,0 @@
/*
* 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.rpc;
import org.apache.dolphinscheduler.common.enums.StateEventType;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceForceStartRequest;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceForceStartResponse;
import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.server.master.event.TaskStateEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.StateEventResponseService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
@Slf4j
@Component
public class TaskInstanceForceStartOperationFunction
implements
ITaskInstanceOperationFunction<TaskInstanceForceStartRequest, TaskInstanceForceStartResponse> {
@Autowired
private StateEventResponseService stateEventResponseService;
@Override
public TaskInstanceForceStartResponse operate(TaskInstanceForceStartRequest taskInstanceForceStartRequest) {
TaskStateEvent stateEvent = TaskStateEvent.builder()
.processInstanceId(taskInstanceForceStartRequest.getProcessInstanceId())
.taskInstanceId(taskInstanceForceStartRequest.getTaskInstanceId())
.key(taskInstanceForceStartRequest.getKey())
.type(StateEventType.WAKE_UP_TASK_GROUP)
.build();
try {
LogUtils.setWorkflowAndTaskInstanceIDMDC(stateEvent.getProcessInstanceId(), stateEvent.getTaskInstanceId());
log.info("Received forceStartTaskInstance, event: {}", stateEvent);
stateEventResponseService.addEvent2WorkflowExecute(stateEvent);
return TaskInstanceForceStartResponse.success();
} finally {
LogUtils.removeWorkflowAndTaskInstanceIdMDC();
}
}
}

View File

@ -17,12 +17,12 @@
package org.apache.dolphinscheduler.server.master.rpc;
import org.apache.dolphinscheduler.common.enums.StateEventType;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceWakeupRequest;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceWakeupResponse;
import org.apache.dolphinscheduler.plugin.task.api.utils.LogUtils;
import org.apache.dolphinscheduler.server.master.event.TaskStateEvent;
import org.apache.dolphinscheduler.server.master.processor.queue.StateEventResponseService;
import org.apache.dolphinscheduler.server.master.cache.ProcessInstanceExecCacheManager;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.WorkflowExecuteRunnable;
import lombok.extern.slf4j.Slf4j;
@ -36,20 +36,30 @@ public class TaskInstanceWakeupOperationFunction
ITaskInstanceOperationFunction<TaskInstanceWakeupRequest, TaskInstanceWakeupResponse> {
@Autowired
private StateEventResponseService stateEventResponseService;
private ProcessInstanceExecCacheManager processInstanceExecCacheManager;
@Override
public TaskInstanceWakeupResponse operate(TaskInstanceWakeupRequest taskInstanceWakeupRequest) {
TaskStateEvent stateEvent = TaskStateEvent.builder()
.processInstanceId(taskInstanceWakeupRequest.getProcessInstanceId())
.taskInstanceId(taskInstanceWakeupRequest.getTaskInstanceId())
.key(taskInstanceWakeupRequest.getKey())
.type(StateEventType.WAKE_UP_TASK_GROUP)
.build();
try {
LogUtils.setWorkflowAndTaskInstanceIDMDC(stateEvent.getProcessInstanceId(), stateEvent.getTaskInstanceId());
log.info("Received wakeupTaskInstance request, event: {}", stateEvent);
stateEventResponseService.addEvent2WorkflowExecute(stateEvent);
log.info("Received TaskInstanceWakeupRequest request{}", taskInstanceWakeupRequest);
int workflowInstanceId = taskInstanceWakeupRequest.getProcessInstanceId();
int taskInstanceId = taskInstanceWakeupRequest.getTaskInstanceId();
LogUtils.setWorkflowAndTaskInstanceIDMDC(workflowInstanceId, taskInstanceId);
WorkflowExecuteRunnable workflowExecuteRunnable =
processInstanceExecCacheManager.getByProcessInstanceId(workflowInstanceId);
if (workflowExecuteRunnable == null) {
log.warn("cannot find WorkflowExecuteRunnable: {}, no need to Wakeup task", workflowInstanceId);
return TaskInstanceWakeupResponse.failed("cannot find WorkflowExecuteRunnable: " + workflowInstanceId);
}
DefaultTaskExecuteRunnable defaultTaskExecuteRunnable =
workflowExecuteRunnable.getTaskExecuteRunnableById(taskInstanceId).orElse(null);
if (defaultTaskExecuteRunnable == null) {
log.warn("Cannot find DefaultTaskExecuteRunnable: {}, cannot Wakeup task", taskInstanceId);
return TaskInstanceWakeupResponse.failed("Cannot find DefaultTaskExecuteRunnable: " + taskInstanceId);
}
defaultTaskExecuteRunnable.dispatch();
log.info("Success Wakeup TaskInstance: {}", taskInstanceId);
return TaskInstanceWakeupResponse.success();
} finally {
LogUtils.removeWorkflowAndTaskInstanceIdMDC();

View File

@ -19,6 +19,8 @@ package org.apache.dolphinscheduler.server.master.rpc;
import org.apache.dolphinscheduler.extract.master.IWorkflowInstanceService;
import org.apache.dolphinscheduler.extract.master.dto.WorkflowExecuteDto;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceWakeupRequest;
import org.apache.dolphinscheduler.extract.master.transportor.TaskInstanceWakeupResponse;
import org.apache.dolphinscheduler.server.master.metrics.ProcessInstanceMetrics;
import org.apache.dolphinscheduler.server.master.service.ExecutingService;
@ -36,6 +38,9 @@ public class WorkflowInstanceServiceImpl implements IWorkflowInstanceService {
@Autowired
private ExecutingService executingService;
@Autowired
private TaskInstanceWakeupOperationFunction taskInstanceWakeupOperationFunction;
@Override
public void clearWorkflowMetrics(Long workflowDefinitionCode) {
log.info("Receive clearWorkflowMetrics request: {}", workflowDefinitionCode);
@ -49,4 +54,9 @@ public class WorkflowInstanceServiceImpl implements IWorkflowInstanceService {
executingService.queryWorkflowExecutingData(workflowInstanceId);
return workflowExecuteDtoOptional.orElse(null);
}
@Override
public TaskInstanceWakeupResponse wakeupTaskInstance(TaskInstanceWakeupRequest taskWakeupRequest) {
return taskInstanceWakeupOperationFunction.operate(taskWakeupRequest);
}
}

Some files were not shown because too many files have changed in this diff Show More