Merge branch 'dev' into dev2

This commit is contained in:
zuo 2024-06-07 13:04:38 +08:00 committed by GitHub
commit 0d35482c44
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
139 changed files with 1641 additions and 1105 deletions

View File

@ -120,6 +120,12 @@ Please refer to the [Quick Start in Kubernetes](../../../docs/docs/en/guide/inst
| conf.auto | bool | `false` | auto restart, if true, all components will be restarted automatically after the common configuration is updated. if false, you need to restart the components manually. default is false |
| conf.common."alert.rpc.port" | int | `50052` | rpc port |
| conf.common."appId.collect" | string | `"log"` | way to collect applicationId: log, aop |
| conf.common."aws.credentials.provider.type" | string | `"AWSStaticCredentialsProvider"` | |
| conf.common."aws.s3.access.key.id" | string | `"minioadmin"` | The AWS access key. if resource.storage.type=S3, and credentials.provider.type is AWSStaticCredentialsProvider. This configuration is required |
| conf.common."aws.s3.access.key.secret" | string | `"minioadmin"` | The AWS secret access key. if resource.storage.type=S3, and credentials.provider.type is AWSStaticCredentialsProvider. This configuration is required |
| conf.common."aws.s3.bucket.name" | string | `"dolphinscheduler"` | The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name. |
| conf.common."aws.s3.endpoint" | string | `"http://minio:9000"` | You need to set this parameter when private cloud s3. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn |
| conf.common."aws.s3.region" | string | `"ca-central-1"` | The AWS Region to use. if resource.storage.type=S3, This configuration is required |
| conf.common."conda.path" | string | `"/opt/anaconda3/etc/profile.d/conda.sh"` | set path of conda.sh |
| 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 |
@ -138,11 +144,6 @@ Please refer to the [Quick Start in Kubernetes](../../../docs/docs/en/guide/inst
| conf.common."resource.alibaba.cloud.oss.bucket.name" | string | `"dolphinscheduler"` | oss bucket name, required if you set resource.storage.type=OSS |
| conf.common."resource.alibaba.cloud.oss.endpoint" | string | `"https://oss-cn-hangzhou.aliyuncs.com"` | oss bucket endpoint, required if you set resource.storage.type=OSS |
| conf.common."resource.alibaba.cloud.region" | string | `"cn-hangzhou"` | alibaba cloud region, required if you set resource.storage.type=OSS |
| conf.common."resource.aws.access.key.id" | string | `"minioadmin"` | The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required |
| conf.common."resource.aws.region" | string | `"ca-central-1"` | The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required |
| conf.common."resource.aws.s3.bucket.name" | string | `"dolphinscheduler"` | The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name. |
| conf.common."resource.aws.s3.endpoint" | string | `"http://minio:9000"` | You need to set this parameter when private cloud s3. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn |
| conf.common."resource.aws.secret.access.key" | string | `"minioadmin"` | The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required |
| conf.common."resource.azure.client.id" | string | `"minioadmin"` | azure storage account name, required if you set resource.storage.type=ABS |
| conf.common."resource.azure.client.secret" | string | `"minioadmin"` | azure storage account key, required if you set resource.storage.type=ABS |
| conf.common."resource.azure.subId" | string | `"minioadmin"` | azure storage subId, required if you set resource.storage.type=ABS |
@ -307,7 +308,6 @@ Please refer to the [Quick Start in Kubernetes](../../../docs/docs/en/guide/inst
| worker.env.WORKER_SERVER_LOAD_PROTECTION_MAX_SYSTEM_CPU_USAGE_PERCENTAGE_THRESHOLDS | float | `0.7` | Worker max system cpu usage, when the worker's system cpu usage is smaller then this value, worker server can be dispatched tasks. |
| worker.env.WORKER_SERVER_LOAD_PROTECTION_MAX_SYSTEM_MEMORY_USAGE_PERCENTAGE_THRESHOLDS | float | `0.7` | Worker max memory usage , when the worker's memory usage is smaller then this value, worker server can be dispatched tasks. |
| worker.env.WORKER_TENANT_CONFIG_AUTO_CREATE_TENANT_ENABLED | bool | `true` | tenant corresponds to the user of the system, which is used by the worker to submit the job. If system does not have this user, it will be automatically created after the parameter worker.tenant.auto.create is true. |
| worker.env.WORKER_TENANT_CONFIG_DISTRIBUTED_TENANT | bool | `false` | Scenes to be used for distributed users. For example, users created by FreeIpa are stored in LDAP. This parameter only applies to Linux, When this parameter is true, worker.tenant.auto.create has no effect and will not automatically create tenants. |
| worker.keda.advanced | object | `{}` | Specify HPA related options |
| worker.keda.cooldownPeriod | int | `30` | How many seconds KEDA will wait before scaling to zero. Note that HPA has a separate cooldown period for scale-downs |
| worker.keda.enabled | bool | `false` | Enable or disable the Keda component |

View File

@ -51,7 +51,6 @@ Create a default common labels.
{{- define "dolphinscheduler.common.labels" -}}
app.kubernetes.io/instance: {{ .Release.Name }}
app.kubernetes.io/managed-by: {{ .Release.Service }}
app.kubernetes.io/version: {{ .Chart.AppVersion }}
{{- end -}}
{{/*

View File

@ -37,7 +37,7 @@ data:
{{- range $key, $value := index .Values.conf "common" }}
{{- if and $.Values.minio.enabled }}
{{- if eq $key "resource.storage.type" }}{{ $value = "S3" }}{{- end }}
{{- if eq $key "resource.aws.s3.endpoint" }}{{ $value = print "http://" (include "dolphinscheduler.minio.fullname" $) ":9000" }}{{- end }}
{{- if eq $key "aws.s3.endpoint" }}{{ $value = print "http://" (include "dolphinscheduler.minio.fullname" $) ":9000" }}{{- end }}
{{- end }}
{{ $key }}={{ $value }}
{{- end }}

View File

@ -250,20 +250,25 @@ conf:
# -- resource store on HDFS/S3 path, resource file will store to this base path, self configuration, please make sure the directory exists on hdfs and have read write permissions. "/dolphinscheduler" is recommended
resource.storage.upload.base.path: /dolphinscheduler
# -- The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id: minioadmin
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
aws.credentials.provider.type: AWSStaticCredentialsProvider
# -- The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key: minioadmin
# -- The AWS access key. if resource.storage.type=S3, and credentials.provider.type is AWSStaticCredentialsProvider. This configuration is required
aws.s3.access.key.id: minioadmin
# -- The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region: ca-central-1
# -- The AWS secret access key. if resource.storage.type=S3, and credentials.provider.type is AWSStaticCredentialsProvider. This configuration is required
aws.s3.access.key.secret: minioadmin
# -- The AWS Region to use. if resource.storage.type=S3, This configuration is required
aws.s3.region: ca-central-1
# -- The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name.
resource.aws.s3.bucket.name: dolphinscheduler
aws.s3.bucket.name: dolphinscheduler
# -- You need to set this parameter when private cloud s3. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn
resource.aws.s3.endpoint: http://minio:9000
aws.s3.endpoint: http://minio:9000
# -- alibaba cloud access key id, required if you set resource.storage.type=OSS
resource.alibaba.cloud.access.key.id: <your-access-key-id>
@ -649,8 +654,6 @@ worker:
WORKER_HOST_WEIGHT: "100"
# -- tenant corresponds to the user of the system, which is used by the worker to submit the job. If system does not have this user, it will be automatically created after the parameter worker.tenant.auto.create is true.
WORKER_TENANT_CONFIG_AUTO_CREATE_TENANT_ENABLED: true
# -- Scenes to be used for distributed users. For example, users created by FreeIpa are stored in LDAP. This parameter only applies to Linux, When this parameter is true, worker.tenant.auto.create has no effect and will not automatically create tenants.
WORKER_TENANT_CONFIG_DISTRIBUTED_TENANT: false
# -- If set true, will use worker bootstrap user as the tenant to execute task when the tenant is `default`;
DEFAULT_TENANT_ENABLED: false

View File

@ -191,19 +191,21 @@ The default configuration is as follows:
Note that DolphinScheduler also supports zookeeper related configuration through `bin/env/dolphinscheduler_env.sh`.
For ETCD Registry, please see more details on [link](https://github.com/apache/dolphinscheduler/blob/dev/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-etcd/README.md).
For JDBC Registry, please see more details on [link](https://github.com/apache/dolphinscheduler/blob/dev/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/README.md).
For ETCD Registry, please see more details
on [link](https://github.com/apache/dolphinscheduler/blob/dev/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-etcd/README.md).
For JDBC Registry, please see more details
on [link](https://github.com/apache/dolphinscheduler/blob/dev/dolphinscheduler-registry/dolphinscheduler-registry-plugins/dolphinscheduler-registry-jdbc/README.md).
### common.properties [hadoop、s3、yarn config properties]
Currently, common.properties mainly configures Hadoop,s3a related configurations. Configuration file location:
| Service | Configuration file |
|---------------|----------------------------------------|
| Master Server | `master-server/conf/common.properties` |
| Api Server | `api-server/conf/common.properties` |
| Worker Server | `worker-server/conf/common.properties` |
| Alert Server | `alert-server/conf/common.properties` |
| Service | Configuration file |
|---------------|-----------------------------------------------------------------------|
| Master Server | `master-server/conf/common.properties` |
| Api Server | `api-server/conf/common.properties`, `api-server/conf/aws.yaml` |
| Worker Server | `worker-server/conf/common.properties`, `worker-server/conf/aws.yaml` |
| Alert Server | `alert-server/conf/common.properties` |
The default configuration is as follows:
@ -212,10 +214,6 @@ The default configuration is as follows:
| data.basedir.path | /tmp/dolphinscheduler | local directory used to store temp files |
| resource.storage.type | NONE | type of resource files: HDFS, S3, OSS, GCS, ABS, NONE |
| resource.upload.path | /dolphinscheduler | storage path of resource files |
| aws.access.key.id | minioadmin | access key id of S3 |
| aws.secret.access.key | minioadmin | secret access key of S3 |
| aws.region | us-east-1 | region of S3 |
| aws.s3.endpoint | http://minio:9000 | endpoint of S3 |
| hdfs.root.user | hdfs | configure users with corresponding permissions if storage type is HDFS |
| fs.defaultFS | hdfs://mycluster:8020 | If resource.storage.type=S3, then the request url would be similar to 's3a://dolphinscheduler'. Otherwise if resource.storage.type=HDFS and hadoop supports HA, copy core-site.xml and hdfs-site.xml into 'conf' directory |
| hadoop.security.authentication.startup.state | false | whether hadoop grant kerberos permission |
@ -330,7 +328,6 @@ Location: `worker-server/conf/application.yaml`
| worker.registry-disconnect-strategy.max-waiting-time | 100s | Used when the worker disconnect from registry, and the disconnect strategy is waiting, this config means the worker will waiting to reconnect to registry in given times, and after the waiting times, if the worker still cannot connect to registry, will stop itself, if the value is 0s, will wait infinitely |
| worker.task-execute-threads-full-policy | REJECT | If REJECT, when the task waiting in the worker reaches exec-threads, it will reject the received task and the Master will redispatch it; If CONTINUE, it will put the task into the worker's execution queue and wait for a free thread to start execution |
| worker.tenant-config.auto-create-tenant-enabled | true | tenant corresponds to the user of the system, which is used by the worker to submit the job. If system does not have this user, it will be automatically created after the parameter worker.tenant.auto.create is true. |
| worker.tenant-config.distributed-tenant-enabled | false | When this parameter is true, auto-create-tenant-enabled has no effect and will not automatically create tenants |
| worker.tenant-config.default-tenant-enabled | false | If set true, will use worker bootstrap user as the tenant to execute task when the tenant is `default`. |
### Alert Server related configuration

View File

@ -153,7 +153,7 @@ The browser access address [http://localhost:5173](http://localhost:5173) can lo
#### zookeeper
Download [ZooKeeper](https://www.apache.org/dyn/closer.lua/zookeeper/zookeeper-3.6.3), and extract it.
Download [ZooKeeper](https://zookeeper.apache.org/releases.html), and extract it.
- Create directory `zkData` and `zkLog`
- Go to the zookeeper installation directory, copy configure file `zoo_sample.cfg` to `conf/zoo.cfg`, and change value of dataDir in conf/zoo.cfg to dataDir=./tmp/zookeeper

View File

@ -14,7 +14,7 @@ Configure all the configurations refer to [pseudo-cluster deployment](pseudo-clu
### Modify Configuration
This step differs quite a lot from [pseudo-cluster deployment](pseudo-cluster.md), because the deployment script transfers the required resources for installation to each deployment machine by using `scp`. So we only need to modify the configuration of the machine that runs `install.sh` script and configurations will dispatch to cluster by `scp`. The configuration file is under the path `bin/env/install_env.sh`, here we only need to modify section **INSTALL MACHINE**, **DolphinScheduler ENV, Database, Registry Server** and keep other sections the same as [pseudo-cluster deployment](pseudo-cluster .md), the following describes the parameters that must be modified:
This step differs quite a lot from [pseudo-cluster deployment](pseudo-cluster.md), because the deployment script transfers the required resources for installation to each deployment machine by using `scp`. So we only need to modify the configuration of the machine that runs `install.sh` script and configurations will dispatch to cluster by `scp`. The configuration file is under the path `bin/env/install_env.sh`, here we only need to modify section **INSTALL MACHINE**, **DolphinScheduler ENV, Database, Registry Server** and keep other sections the same as [pseudo-cluster deployment](pseudo-cluster.md), the following describes the parameters that must be modified:
```shell
# ---------------------------------------------------------

View File

@ -2,7 +2,7 @@
The purpose of the pseudo-cluster deployment is to deploy the DolphinScheduler service on a single machine. In this mode, DolphinScheduler's master, worker, API server, are all on the same machine.
If you are a new hand and want to experience DolphinScheduler functions, we recommend you install follow [Standalone deployment](standalone.md). If you want to experience more complete functions and schedule massive tasks, we recommend you install follow[pseudo-cluster deployment. If you want to deploy DolphinScheduler in production, we recommend you follow [cluster deployment](cluster.md) or [Kubernetes deployment](kubernetes.md).
If you are a new hand and want to experience DolphinScheduler functions, we recommend you install follow [Standalone deployment](standalone.md). If you want to experience more complete functions and schedule massive tasks, we recommend you install follow [pseudo-cluster deployment](pseudo-cluster.md). If you want to deploy DolphinScheduler in production, we recommend you follow [cluster deployment](cluster.md) or [Kubernetes deployment](kubernetes.md).
## Preparation

View File

@ -36,7 +36,7 @@ Access address `http://localhost:12345/dolphinscheduler/ui` and login DolphinSch
### Start or Stop Server
The script `./bin/dolphinscheduler-daemon.sh`can be used not only quickly start standalone, but also to stop the service operation. The following are all the commands:
The script `./bin/dolphinscheduler-daemon.sh` can be used not only quickly start standalone, but also to stop the service operation. The following are all the commands:
```shell
# Start Standalone Server

View File

@ -28,74 +28,37 @@ The configuration you may need to change:
## connect AWS S3
if you want to upload resources to `Resource Center` connected to `S3`, you need to configure `api-server/conf/common.properties` and `worker-server/conf/common.properties`. You can refer to the following:
if you want to upload resources to `Resource Center` connected to `S3`, you need to configure `api-server/conf/common.properties`, `api-server/conf/aws.yaml` and `worker-server/conf/common.properties`, `worker-server/conf/aws.yaml`. You can refer to the following:
config the following fields
```properties
......
resource.storage.type=S3
......
resource.aws.access.key.id=aws_access_key_id
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=aws_secret_access_key
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=us-west-2
# The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name.
resource.aws.s3.bucket.name=dolphinscheduler
# You need to set this parameter when private cloud s4. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn
resource.aws.s3.endpoint=
......
```
## Use HDFS or Remote Object Storage
```yaml
aws:
s3:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: <access.key.id>
access.key.secret: <access.key.secret>
region: <region>
bucket.name: <bucket.name>
endpoint: <endpoint>
After version 3.0.0-alpha, if you want to upload resources to `Resource Center` connected to `HDFS`, you need to configure `api-server/conf/common.properties` and `worker-server/conf/common.properties`.
```
## connect OSS S3
if you want to upload resources to `Resource Center` connected to `OSS`, you need to configure `api-server/conf/common.properties` and `worker-server/conf/common.properties`. You can refer to the following:
config the following fields
```properties
#
# 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.
#
# user data local directory path, please make sure the directory exists and have read write permissions
data.basedir.path=/tmp/dolphinscheduler
# resource view suffixs
#resource.view.suffixs=txt,log,sh,bat,conf,cfg,py,java,sql,xml,hql,properties,json,yml,yaml,ini,js
# resource storage type: LOCAL, HDFS, S3, OSS, GCS, ABS, OBS
resource.storage.type=LOCAL
# resource store on HDFS/S3/OSS path, resource file will store to this base path, self configuration, please make sure the directory exists on hdfs and have read write permissions. "/dolphinscheduler" is recommended
resource.storage.upload.base.path=/tmp/dolphinscheduler
# The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id=minioadmin
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=minioadmin
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=cn-north-1
# The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name.
resource.aws.s3.bucket.name=dolphinscheduler
# You need to set this parameter when private cloud s3. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn
resource.aws.s3.endpoint=http://localhost:9000
# alibaba cloud access key id, required if you set resource.storage.type=OSS
resource.alibaba.cloud.access.key.id=<your-access-key-id>
# alibaba cloud access key secret, required if you set resource.storage.type=OSS
@ -107,89 +70,24 @@ resource.alibaba.cloud.oss.bucket.name=dolphinscheduler
# oss bucket endpoint, required if you set resource.storage.type=OSS
resource.alibaba.cloud.oss.endpoint=https://oss-cn-hangzhou.aliyuncs.com
# alibaba cloud access key id, required if you set resource.storage.type=OBS
```
## connect OBS S3
if you want to upload resources to `Resource Center` connected to `OBS`, you need to configure `api-server/conf/common.properties` and `worker-server/conf/common.properties`. You can refer to the following:
config the following fields
```properties
# access key id, required if you set resource.storage.type=OBS
resource.huawei.cloud.access.key.id=<your-access-key-id>
# alibaba cloud access key secret, required if you set resource.storage.type=OBS
# access key secret, required if you set resource.storage.type=OBS
resource.huawei.cloud.access.key.secret=<your-access-key-secret>
# oss bucket name, required if you set resource.storage.type=OBS
resource.huawei.cloud.obs.bucket.name=dolphinscheduler
# oss bucket endpoint, required if you set resource.storage.type=OBS
resource.huawei.cloud.obs.endpoint=obs.cn-southwest-2.huaweicloud.com
# if resource.storage.type=HDFS, the user must have the permission to create directories under the HDFS root path
resource.hdfs.root.user=hdfs
# if resource.storage.type=S3, the value like: s3a://dolphinscheduler; if resource.storage.type=HDFS and namenode HA is enabled, you need to copy core-site.xml and hdfs-site.xml to conf dir
resource.hdfs.fs.defaultFS=hdfs://mycluster:8020
# whether to startup kerberos
hadoop.security.authentication.startup.state=false
# java.security.krb5.conf path
java.security.krb5.conf.path=/opt/krb5.conf
# login user from keytab username
login.user.keytab.username=hdfs-mycluster@ESZ.COM
# login user from keytab path
login.user.keytab.path=/opt/hdfs.headless.keytab
# kerberos expire time, the unit is hour
kerberos.expire.time=2
# resourcemanager port, the default value is 8088 if not specified
resource.manager.httpaddress.port=8088
# if resourcemanager HA is enabled, please set the HA IPs; if resourcemanager is single, keep this value empty
yarn.resourcemanager.ha.rm.ids=192.168.xx.xx,192.168.xx.xx
# if resourcemanager HA is enabled or not use resourcemanager, please keep the default value; If resourcemanager is single, you only need to replace ds1 to actual resourcemanager hostname
yarn.application.status.address=http://ds1:%s/ws/v1/cluster/apps/%s
# job history status url when application number threshold is reached(default 10000, maybe it was set to 1000)
yarn.job.history.status.address=http://ds1:19888/ws/v1/history/mapreduce/jobs/%s
# datasource encryption enable
datasource.encryption.enable=false
# datasource encryption salt
datasource.encryption.salt=!@#$%^&*
# 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
# Network IP gets priority, default inner outer
# Whether hive SQL is executed in the same session
support.hive.oneSession=false
# use sudo or not, if set true, executing user is tenant user and deploy user needs sudo permissions; if set false, executing user is the deploy user and doesn't need sudo permissions
sudo.enable=true
# network interface preferred like eth0, default: empty
#dolphin.scheduler.network.interface.preferred=
# network IP gets priority, default: inner outer
#dolphin.scheduler.network.priority.strategy=default
# system env path
#dolphinscheduler.env.path=dolphinscheduler_env.sh
# development state
development.state=false
# rpc port
alert.rpc.port=50052
# set path of conda.sh
conda.path=/opt/anaconda3/etc/profile.d/conda.sh
# Task resource limit state
task.resource.limit.state=false
# way to collect applicationId: log(original regex match), aop
appId.collect: log
```
> **Note:**

View File

@ -73,14 +73,17 @@ Parameters of restarting the task by interface
## Environment to prepare
Some AWS configuration is required, modify a field in file `common.properties`
Some AWS configuration is required, modify a field in file `aws.yaml`
```yaml
# The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id=<YOUR AWS ACCESS KEY>
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=<YOUR AWS SECRET KEY>
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=<AWS REGION>
dms:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: <access.key.id>
access.key.secret: <access.key.secret>
region: <region>
endpoint: <endpoint>
```

View File

@ -35,14 +35,17 @@ The task plugin are shown as follows:
## Environment to prepare
Some AWS configuration is required, modify a field in file `common.properties`
Some AWS configuration is required, modify a field in file `aws.yaml`
```yaml
# The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id=<YOUR AWS ACCESS KEY>
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=<YOUR AWS SECRET KEY>
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=<AWS REGION>
sagemaker:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: <access.key.id>
access.key.secret: <access.key.secret>
region: <region>
endpoint: <endpoint>
```

View File

@ -331,7 +331,6 @@ common.properties配置文件目前主要是配置hadoop/s3/yarn/applicationId
| worker.registry-disconnect-strategy.max-waiting-time | 100s | 当Worker与注册中心失联之后重连时间, 之后当strategy为waiting时该值生效。 该值表示当Worker与注册中心失联时会在给定时间之内进行重连, 在给定时间之内重连失败将会停止自己在重连时Worker会丢弃kill正在执行的任务。值为0表示会无限期等待 |
| worker.task-execute-threads-full-policy | REJECT | 如果是 REJECT, 当Worker中等待队列中的任务数达到exec-threads时, Worker将会拒绝接下来新接收的任务Master将会重新分发该任务; 如果是 CONTINUE, Worker将会接收任务放入等待队列中等待空闲线程去执行该任务 |
| worker.tenant-config.auto-create-tenant-enabled | true | 租户对应于系统的用户,由worker提交作业.如果系统没有该用户,则在参数worker.tenant.auto.create为true后自动创建。 |
| worker.tenant-config.distributed-tenant-enabled | false | 如果设置为true, auto-create-tenant-enabled 将会不起作用。 |
| worker.tenant-config.default-tenant-enabled | false | 如果设置为true, 将会使用worker服务启动用户作为 `default` 租户。 |
## Alert Server相关配置

View File

@ -148,7 +148,7 @@ pnpm run dev
#### zookeeper
下载 [ZooKeeper](https://www.apache.org/dyn/closer.lua/zookeeper/zookeeper-3.6.3),解压
下载 [ZooKeeper](https://zookeeper.apache.org/releases.html),解压
* 在 ZooKeeper 的目录下新建 zkData、zkLog文件夹
* 将 conf 目录下的 `zoo_sample.cfg` 文件,复制一份,重命名为 `zoo.cfg`,修改其中数据和日志的配置,如:

View File

@ -26,77 +26,35 @@ Dolphinscheduler 资源中心使用本地系统默认是开启的,不需要用
## 对接AWS S3
如果需要使用到资源中心的 S3 上传资源,我们需要对以下路径的进行配置:`api-server/conf/common.properties` 和 `worker-server/conf/common.properties`。可参考如下:
如果需要使用到资源中心的 S3 上传资源,我们需要对以下路径的进行配置:`api-server/conf/common.properties`, `api-server/conf/aws.yaml``worker-server/conf/common.properties`, `worker-server/conf/aws.yaml`。可参考如下:
配置以下字段
```properties
......
resource.storage.type=S3
......
resource.aws.access.key.id=aws_access_key_id
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=aws_secret_access_key
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=us-west-2
# The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name.
resource.aws.s3.bucket.name=dolphinscheduler
# You need to set this parameter when private cloud s4. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn
resource.aws.s3.endpoint=
......
```
## 对接分布式或远端对象存储
```yaml
aws:
s3:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: <access.key.id>
access.key.secret: <access.key.secret>
region: <region>
bucket.name: <bucket.name>
endpoint: <endpoint>
当需要使用资源中心进行相关文件的创建或者上传操作时,所有的文件和资源都会被存储在分布式文件系统`HDFS`或者远端的对象存储,如`S3`上。所以需要进行以下配置:
```
### 配置 common.properties 文件
## 对接阿里云 OSS
在 3.0.0-alpha 版本之后,如果需要使用到资源中心的 HDFS 或 S3 上传资源,我们需要对以下路径的进行配置:`api-server/conf/common.properties` 和 `worker-server/conf/common.properties`。可参考如下:
如果需要使用到资源中心的 OSS 上传资源,我们需要对以下路径的进行配置:`api-server/conf/common.properties` 和 `worker-server/conf/common.properties`。可参考如下:
```properties
#
# 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.
#
# user data local directory path, please make sure the directory exists and have read write permissions
data.basedir.path=/tmp/dolphinscheduler
# resource storage type: LOCAL, HDFS, S3, OSS, GCS, ABS, OBS
resource.storage.type=LOCAL
# resource store on HDFS/S3/OSS path, resource file will store to this hadoop hdfs path, self configuration,
# please make sure the directory exists on hdfs and have read write permissions. "/dolphinscheduler" is recommended
resource.storage.upload.base.path=/tmp/dolphinscheduler
# The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id=minioadmin
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=minioadmin
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=cn-north-1
# The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name.
resource.aws.s3.bucket.name=dolphinscheduler
# You need to set this parameter when private cloud s3. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn
resource.aws.s3.endpoint=http://localhost:9000
# alibaba cloud access key id, required if you set resource.storage.type=OSS
resource.alibaba.cloud.access.key.id=<your-access-key-id>
# alibaba cloud access key secret, required if you set resource.storage.type=OSS
@ -108,87 +66,22 @@ resource.alibaba.cloud.oss.bucket.name=dolphinscheduler
# oss bucket endpoint, required if you set resource.storage.type=OSS
resource.alibaba.cloud.oss.endpoint=https://oss-cn-hangzhou.aliyuncs.com
# alibaba cloud access key id, required if you set resource.storage.type=OBS
```
## 对接华为云 OBS
如果需要使用到资源中心的 OBS 上传资源,我们需要对以下路径的进行配置:`api-server/conf/common.properties` 和 `worker-server/conf/common.properties`。可参考如下:
```properties
# access key id, required if you set resource.storage.type=OBS
resource.huawei.cloud.access.key.id=<your-access-key-id>
# alibaba cloud access key secret, required if you set resource.storage.type=OBS
# access key secret, required if you set resource.storage.type=OBS
resource.huawei.cloud.access.key.secret=<your-access-key-secret>
# oss bucket name, required if you set resource.storage.type=OBS
resource.huawei.cloud.obs.bucket.name=dolphinscheduler
# oss bucket endpoint, required if you set resource.storage.type=OBS
resource.huawei.cloud.obs.endpoint=obs.cn-southwest-2.huaweicloud.com
# if resource.storage.type=HDFS, the user must have the permission to create directories under the HDFS root path
resource.hdfs.root.user=root
# if resource.storage.type=S3, the value like: s3a://dolphinscheduler;
# if resource.storage.type=HDFS and namenode HA is enabled, you need to copy core-site.xml and hdfs-site.xml to conf dir
resource.hdfs.fs.defaultFS=hdfs://localhost:8020
# whether to startup kerberos
hadoop.security.authentication.startup.state=false
# java.security.krb5.conf path
java.security.krb5.conf.path=/opt/krb5.conf
# login user from keytab username
login.user.keytab.username=hdfs-mycluster@ESZ.COM
# login user from keytab path
login.user.keytab.path=/opt/hdfs.headless.keytab
# kerberos expire time, the unit is hour
kerberos.expire.time=2
# resource view suffixs
#resource.view.suffixs=txt,log,sh,bat,conf,cfg,py,java,sql,xml,hql,properties,json,yml,yaml,ini,js
# resourcemanager port, the default value is 8088 if not specified
resource.manager.httpaddress.port=8088
# if resourcemanager HA is enabled, please set the HA IPs; if resourcemanager is single, keep this value empty
yarn.resourcemanager.ha.rm.ids=192.168.xx.xx,192.168.xx.xx
# if resourcemanager HA is enabled or not use resourcemanager, please keep the default value;
# If resourcemanager is single, you only need to replace ds1 to actual resourcemanager hostname
yarn.application.status.address=http://localhost:%s/ds/v1/cluster/apps/%s
# job history status url when application number threshold is reached(default 10000, maybe it was set to 1000)
yarn.job.history.status.address=http://localhost:19888/ds/v1/history/mapreduce/jobs/%s
# datasource encryption enable
datasource.encryption.enable=false
# datasource encryption salt
datasource.encryption.salt=!@#$%^&*
# 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
# Network IP gets priority, default inner outer
# Whether hive SQL is executed in the same session
support.hive.oneSession=false
# use sudo or not, if set true, executing user is tenant user and deploy user needs sudo permissions;
# if set false, executing user is the deploy user and doesn't need sudo permissions
sudo.enable=true
# network interface preferred like eth0, default: empty
#dolphin.scheduler.network.interface.preferred=
# network IP gets priority, default: inner outer
#dolphin.scheduler.network.priority.strategy=default
# system env path
#dolphinscheduler.env.path=env/dolphinscheduler_env.sh
# development state
development.state=false
# rpc port
alert.rpc.port=50052
# way to collect applicationId: log(original regex match), aop
appId.collect: log
```
> **注意**

View File

@ -73,14 +73,17 @@ DolphinScheduler 在 启动DMS 任务后会跟中DMS任务状态直至DMS
## 环境配置
需要进行AWS的一些配置修改`common.properties`中的以下配置信息
需要进行AWS的一些配置修改`aws.yml`中的以下配置信息
```yaml
# The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id=<YOUR AWS ACCESS KEY>
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=<YOUR AWS SECRET KEY>
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=<AWS REGION>
dms:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: <access.key.id>
access.key.secret: <access.key.secret>
region: <region>
endpoint: <endpoint>
```

View File

@ -33,14 +33,17 @@ DolphinScheduler SageMaker 组件的功能:
## 环境配置
需要进行AWS的一些配置修改`common.properties`中的`xxxxx`为你的配置信息
需要进行AWS的一些配置修改`aws.yml`中的以下配置信息
```yaml
# The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id=<YOUR AWS ACCESS KEY>
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=<YOUR AWS SECRET KEY>
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=<AWS REGION>
sagemaker:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: <access.key.id>
access.key.secret: <access.key.secret>
region: <region>
endpoint: <endpoint>
```

View File

@ -56,6 +56,13 @@
</includes>
<outputDirectory>conf</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-authentication/dolphinscheduler-aws-authentication/src/main/resources</directory>
<includes>
<include>**/*.yaml</include>
</includes>
<outputDirectory>conf</outputDirectory>
</fileSet>
</fileSets>
<dependencySets>
<dependencySet>

View File

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

View File

@ -57,6 +57,13 @@
</includes>
<outputDirectory>conf</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-authentication/dolphinscheduler-aws-authentication/src/main/resources</directory>
<includes>
<include>**/*.yaml</include>
</includes>
<outputDirectory>conf</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-ui/dist</directory>
<outputDirectory>./ui</outputDirectory>

View File

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

View File

@ -126,7 +126,7 @@ public class OperatorUtils {
return auditType.getAuditOperationType();
}
public static long getObjectIdentityByParma(String[] paramNameArr, Map<String, Object> paramsMap) {
public static long getObjectIdentityByParam(String[] paramNameArr, Map<String, Object> paramsMap) {
for (String name : paramNameArr) {
if (paramsMap.get(name) instanceof String) {
String param = (String) paramsMap.get(name);

View File

@ -80,14 +80,14 @@ public abstract class BaseAuditOperator implements AuditOperator {
}
modifyRequestParams(paramNameArr, paramsMap, auditLogList);
setObjectByParma(paramNameArr, paramsMap, auditLogList);
setObjectByParam(paramNameArr, paramsMap, auditLogList);
if (auditLogList.get(0).getModelId() == null) {
auditLogList.get(0).setModelId(OperatorUtils.getObjectIdentityByParma(paramNameArr, paramsMap));
auditLogList.get(0).setModelId(OperatorUtils.getObjectIdentityByParam(paramNameArr, paramsMap));
}
}
protected void setObjectByParma(String[] paramNameArr, Map<String, Object> paramsMap,
protected void setObjectByParam(String[] paramNameArr, Map<String, Object> paramsMap,
List<AuditLog> auditLogList) {
String name = paramNameArr[0];
@ -97,7 +97,7 @@ public abstract class BaseAuditOperator implements AuditOperator {
return;
}
String objName = getObjectNameFromReturnIdentity(value);
String objName = getObjectNameFromIdentity(value);
if (Strings.isNullOrEmpty(objName)) {
auditLogList.get(0).setModelName(value.toString());
@ -114,7 +114,7 @@ public abstract class BaseAuditOperator implements AuditOperator {
auditLogList.get(0).setModelName(objName);
}
protected void setObjectByParmaArr(String[] paramNameArr, Map<String, Object> paramsMap,
protected void setObjectByParamArr(String[] paramNameArr, Map<String, Object> paramsMap,
List<AuditLog> auditLogList) {
AuditLog auditLog = auditLogList.get(0);
@ -127,7 +127,7 @@ public abstract class BaseAuditOperator implements AuditOperator {
for (String identityString : identityArr) {
long identity = toLong(identityString);
String value = getObjectNameFromReturnIdentity(identity);
String value = getObjectNameFromIdentity(identity);
if (value == null) {
continue;
@ -156,7 +156,7 @@ public abstract class BaseAuditOperator implements AuditOperator {
protected void setObjectNameFromReturnIdentity(List<AuditLog> auditLogList) {
auditLogList
.forEach(auditLog -> auditLog.setModelName(getObjectNameFromReturnIdentity(auditLog.getModelId())));
.forEach(auditLog -> auditLog.setModelName(getObjectNameFromIdentity(auditLog.getModelId())));
}
protected void modifyObjectFromReturnObject(String[] params, Map<String, Object> returnObjectMap,
@ -180,7 +180,7 @@ public abstract class BaseAuditOperator implements AuditOperator {
return NumberUtils.toLong(str.toString(), -1);
}
protected String getObjectNameFromReturnIdentity(Object identity) {
protected String getObjectNameFromIdentity(Object identity) {
return identity.toString();
}

View File

@ -31,7 +31,7 @@ public class AlertGroupAuditOperatorImpl extends BaseAuditOperator {
private AlertGroupMapper alertGroupMapper;
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -31,7 +31,7 @@ public class AlertInstanceAuditOperatorImpl extends BaseAuditOperator {
private AlertPluginInstanceMapper alertPluginInstanceMapper;
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -31,7 +31,7 @@ public class ClusterAuditOperatorImpl extends BaseAuditOperator {
private ClusterMapper clusterMapper;
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -31,7 +31,7 @@ public class DatasourceAuditOperatorImpl extends BaseAuditOperator {
private DataSourceMapper dataSourceMapper;
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -31,7 +31,7 @@ public class EnvironmentAuditOperatorImpl extends BaseAuditOperator {
private EnvironmentMapper environmentMapper;
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -31,7 +31,7 @@ public class K8SNamespaceAuditOperatorImpl extends BaseAuditOperator {
private K8sNamespaceMapper k8sNamespaceMapper;
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -46,14 +46,14 @@ public class ProcessAuditOperatorImpl extends BaseAuditOperator {
}
@Override
protected void setObjectByParma(String[] paramNameArr, Map<String, Object> paramsMap,
protected void setObjectByParam(String[] paramNameArr, Map<String, Object> paramsMap,
List<AuditLog> auditLogList) {
if (paramNameArr[0].equals(AuditLogConstants.CODES)
|| paramNameArr[0].equals(AuditLogConstants.PROCESS_DEFINITION_CODES)
|| paramNameArr[0].equals(AuditLogConstants.PROCESS_INSTANCE_IDS)) {
super.setObjectByParmaArr(paramNameArr, paramsMap, auditLogList);
super.setObjectByParamArr(paramNameArr, paramsMap, auditLogList);
} else {
super.setObjectByParma(paramNameArr, paramsMap, auditLogList);
super.setObjectByParam(paramNameArr, paramsMap, auditLogList);
}
if (paramsMap.containsKey(AuditLogConstants.VERSION)) {
if (paramsMap.get(AuditLogConstants.VERSION) != null) {
@ -65,7 +65,7 @@ public class ProcessAuditOperatorImpl extends BaseAuditOperator {
}
@Override
protected String getObjectNameFromReturnIdentity(Object identity) {
protected String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -48,17 +48,17 @@ public class ProcessInstanceAuditOperatorImpl extends BaseAuditOperator {
}
@Override
protected void setObjectByParma(String[] paramNameArr, Map<String, Object> paramsMap,
protected void setObjectByParam(String[] paramNameArr, Map<String, Object> paramsMap,
List<AuditLog> auditLogList) {
if (paramNameArr[0].equals(AuditLogConstants.PROCESS_INSTANCE_IDS)) {
super.setObjectByParmaArr(paramNameArr, paramsMap, auditLogList);
super.setObjectByParamArr(paramNameArr, paramsMap, auditLogList);
} else {
super.setObjectByParma(paramNameArr, paramsMap, auditLogList);
super.setObjectByParam(paramNameArr, paramsMap, auditLogList);
}
}
@Override
protected String getObjectNameFromReturnIdentity(Object identity) {
protected String getObjectNameFromIdentity(Object identity) {
int objId = NumberUtils.toInt(identity.toString(), -1);
if (objId == -1) {
return "";

View File

@ -34,7 +34,7 @@ public class ProjectAuditOperatorImpl extends BaseAuditOperator {
private ProjectMapper projectMapper;
@Override
protected String getObjectNameFromReturnIdentity(Object identity) {
protected String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -37,7 +37,7 @@ public class ResourceAuditOperatorImpl extends BaseAuditOperator {
}
@Override
protected void setObjectByParma(String[] paramNameArr, Map<String, Object> paramsMap,
protected void setObjectByParam(String[] paramNameArr, Map<String, Object> paramsMap,
List<AuditLog> auditLogList) {
Object objName = getFileNameFromParam(paramNameArr, paramsMap);

View File

@ -72,7 +72,7 @@ public class ScheduleAuditOperatorImpl extends BaseAuditOperator {
}
@Override
protected String getObjectNameFromReturnIdentity(Object identity) {
protected String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -46,17 +46,17 @@ public class TaskAuditOperatorImpl extends BaseAuditOperator {
}
@Override
protected void setObjectByParma(String[] paramNameArr, Map<String, Object> paramsMap,
protected void setObjectByParam(String[] paramNameArr, Map<String, Object> paramsMap,
List<AuditLog> auditLogList) {
super.setObjectByParma(paramNameArr, paramsMap, auditLogList);
super.setObjectByParam(paramNameArr, paramsMap, auditLogList);
if (paramsMap.containsKey(AuditLogConstants.VERSION)) {
auditLogList.get(0).setDetail(paramsMap.get(AuditLogConstants.VERSION).toString());
}
}
@Override
protected String getObjectNameFromReturnIdentity(Object identity) {
protected String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -31,7 +31,7 @@ public class TaskGroupAuditOperatorImpl extends BaseAuditOperator {
private TaskGroupMapper taskGroupMapper;
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -31,7 +31,7 @@ public class TaskInstancesAuditOperatorImpl extends BaseAuditOperator {
private TaskInstanceMapper taskInstanceMapper;
@Override
protected String getObjectNameFromReturnIdentity(Object identity) {
protected String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -31,7 +31,7 @@ public class TenantAuditOperatorImpl extends BaseAuditOperator {
private TenantMapper tenantMapper;
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -54,7 +54,7 @@ public class TokenAuditOperatorImpl extends BaseAuditOperator {
}
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -33,7 +33,7 @@ public class UdfFunctionAuditOperatorImpl extends BaseAuditOperator {
private UdfFuncMapper udfFuncMapper;
@Override
protected String getObjectNameFromReturnIdentity(Object identity) {
protected String getObjectNameFromIdentity(Object identity) {
int objId = NumberUtils.toInt(identity.toString(), -1);
if (objId == -1) {
return "";

View File

@ -31,7 +31,7 @@ public class UserAuditOperatorImpl extends BaseAuditOperator {
private UserMapper userMapper;
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -48,7 +48,7 @@ public class WorkerGroupAuditOperatorImpl extends BaseAuditOperator {
}
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -31,7 +31,7 @@ public class YarnQueueAuditOperatorImpl extends BaseAuditOperator {
private QueueMapper queueMapper;
@Override
public String getObjectNameFromReturnIdentity(Object identity) {
public String getObjectNameFromIdentity(Object identity) {
Long objId = toLong(identity);
if (objId == -1) {
return "";

View File

@ -44,7 +44,7 @@ public class ApiExceptionHandler {
@ExceptionHandler(Throwable.class)
public Result<Object> exceptionHandler(Throwable e, HandlerMethod hm) {
ApiException ce = hm.getMethodAnnotation(ApiException.class);
log.error("Meet en unknown exception: ", e);
log.error("Meet an unknown exception: ", e);
if (ce == null) {
return Result.errorWithArgs(Status.INTERNAL_SERVER_ERROR_ARGS, e.getMessage());
}

View File

@ -17,7 +17,6 @@
package org.apache.dolphinscheduler.api.service.impl;
import static org.apache.dolphinscheduler.api.utils.CheckUtils.checkFilePath;
import static org.apache.dolphinscheduler.common.constants.Constants.ALIAS;
import static org.apache.dolphinscheduler.common.constants.Constants.CONTENT;
import static org.apache.dolphinscheduler.common.constants.Constants.EMPTY_STRING;
@ -1291,10 +1290,6 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
if (FOLDER_SEPARATOR.equalsIgnoreCase(fullName)) {
return;
}
// abnormal characters check
if (!checkFilePath(fullName)) {
throw new ServiceException(Status.ILLEGAL_RESOURCE_PATH);
}
// Avoid returning to the parent directory
if (fullName.contains("../")) {
throw new ServiceException(Status.ILLEGAL_RESOURCE_PATH, fullName);

View File

@ -158,14 +158,4 @@ public class CheckUtils {
return pattern.matcher(str).matches();
}
/**
* regex FilePath check,only use a to z, A to Z, 0 to 9, and _./-
*
* @param str input string
* @return true if regex pattern is right, otherwise return false
*/
public static boolean checkFilePath(String str) {
return regexChecks(str, Constants.REGEX_FILE_PATH);
}
}

View File

@ -92,24 +92,4 @@ public class CheckUtilsTest {
Assertions.assertTrue(CheckUtils.checkPhone("17362537263"));
}
/**
* check file path
*/
@Test
public void testCheckFilePath() {
// true
Assertions.assertTrue(CheckUtils.checkFilePath("/"));
Assertions.assertTrue(CheckUtils.checkFilePath("xx/"));
Assertions.assertTrue(CheckUtils.checkFilePath("/xx"));
Assertions.assertTrue(CheckUtils.checkFilePath("14567134578654"));
Assertions.assertTrue(CheckUtils.checkFilePath("/admin/root/"));
Assertions.assertTrue(CheckUtils.checkFilePath("/admin/root/1531531..13513/153135.."));
// false
Assertions.assertFalse(CheckUtils.checkFilePath(null));
Assertions.assertFalse(CheckUtils.checkFilePath("file://xxx/ss"));
Assertions.assertFalse(CheckUtils.checkFilePath("/xxx/ss;/dasd/123"));
Assertions.assertFalse(CheckUtils.checkFilePath("/xxx/ss && /dasd/123"));
Assertions.assertFalse(CheckUtils.checkFilePath("/xxx/ss || /dasd/123"));
}
}

View File

@ -0,0 +1,64 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Licensed to the Apache Software Foundation (ASF) under one or more
~ contributor license agreements. See the NOTICE file distributed with
~ this work for additional information regarding copyright ownership.
~ The ASF licenses this file to You under the Apache License, Version 2.0
~ (the "License"); you may not use this file except in compliance with
~ the License. You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-authentication</artifactId>
<version>dev-SNAPSHOT</version>
</parent>
<artifactId>dolphinscheduler-aws-authentication</artifactId>
<dependencies>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-emr</artifactId>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-s3</artifactId>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-sagemaker</artifactId>
</dependency>
<dependency>
<groupId>com.amazonaws</groupId>
<artifactId>aws-java-sdk-dms</artifactId>
</dependency>
<dependency>
<groupId>software.amazon.awssdk</groupId>
<artifactId>datasync</artifactId>
</dependency>
<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
<scope>provided</scope>
</dependency>
</dependencies>
</project>

View File

@ -0,0 +1,74 @@
/*
* 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.authentication.aws;
import static org.apache.dolphinscheduler.authentication.aws.AwsConfigurationKeys.AWS_AUTHENTICATION_TYPE;
import java.util.Map;
import lombok.experimental.UtilityClass;
import lombok.extern.slf4j.Slf4j;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.auth.InstanceProfileCredentialsProvider;
@Slf4j
@UtilityClass
public class AWSCredentialsProviderFactor {
public static AWSCredentialsProvider credentialsProvider(Map<String, String> awsProperties) {
String awsAuthenticationType = awsProperties.getOrDefault(
AWS_AUTHENTICATION_TYPE, AWSCredentialsProviderType.STATIC_CREDENTIALS_PROVIDER.getName());
AWSCredentialsProviderType awsCredentialsProviderType =
AWSCredentialsProviderType.of(awsAuthenticationType).orElse(null);
if (awsCredentialsProviderType == null) {
throw new IllegalArgumentException(
"The aws.credentials.provider.type: " + awsAuthenticationType + " is invalidated");
}
switch (awsCredentialsProviderType) {
case STATIC_CREDENTIALS_PROVIDER:
return createAWSStaticCredentialsProvider(awsProperties);
case INSTANCE_PROFILE_CREDENTIALS_PROVIDER:
return createInstanceProfileCredentialsProvider();
default:
throw new IllegalArgumentException(
"The aws.credentials.provider.type: " + awsAuthenticationType + " is invalidated");
}
}
private static AWSCredentialsProvider createAWSStaticCredentialsProvider(Map<String, String> awsProperties) {
String awsAccessKeyId = awsProperties.get(AwsConfigurationKeys.AWS_ACCESS_KEY_ID);
String awsSecretAccessKey = awsProperties.get(AwsConfigurationKeys.AWS_SECRET);
final BasicAWSCredentials basicAWSCredentials = new BasicAWSCredentials(awsAccessKeyId, awsSecretAccessKey);
AWSStaticCredentialsProvider awsStaticCredentialsProvider =
new AWSStaticCredentialsProvider(basicAWSCredentials);
log.info("AWSStaticCredentialsProvider created successfully");
return awsStaticCredentialsProvider;
}
private static AWSCredentialsProvider createInstanceProfileCredentialsProvider() {
InstanceProfileCredentialsProvider instanceProfileCredentialsProvider =
InstanceProfileCredentialsProvider.getInstance();
log.info("InstanceProfileCredentialsProvider created successfully");
return instanceProfileCredentialsProvider;
}
}

View File

@ -0,0 +1,49 @@
/*
* 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.authentication.aws;
import java.util.Optional;
import lombok.Getter;
@Getter
public enum AWSCredentialsProviderType {
STATIC_CREDENTIALS_PROVIDER("AWSStaticCredentialsProvider"),
INSTANCE_PROFILE_CREDENTIALS_PROVIDER("InstanceProfileCredentialsProvider"),
;
private final String name;
AWSCredentialsProviderType(String name) {
this.name = name;
}
public static Optional<AWSCredentialsProviderType> of(String name) {
if (name == null) {
return Optional.empty();
}
for (AWSCredentialsProviderType type : values()) {
if (type.getName().equalsIgnoreCase(name)) {
return Optional.of(type);
}
}
return Optional.empty();
}
}

View File

@ -0,0 +1,53 @@
/*
* 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.authentication.aws;
import java.util.Map;
import lombok.experimental.UtilityClass;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.databasemigrationservice.AWSDatabaseMigrationService;
import com.amazonaws.services.databasemigrationservice.AWSDatabaseMigrationServiceClientBuilder;
@UtilityClass
public class AWSDatabaseMigrationServiceClientFactory {
public AWSDatabaseMigrationService createAWSDatabaseMigrationServiceClient(Map<String, String> awsProperties) {
AWSCredentialsProvider awsCredentialsProvider = AWSCredentialsProviderFactor.credentialsProvider(awsProperties);
Regions regions = Regions.fromName(awsProperties.get(AwsConfigurationKeys.AWS_REGION));
String endpoint = awsProperties.get(AwsConfigurationKeys.AWS_ENDPOINT);
if (endpoint != null && !endpoint.isEmpty()) {
return AWSDatabaseMigrationServiceClientBuilder
.standard()
.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint, regions.getName()))
.withCredentials(awsCredentialsProvider)
.build();
} else {
return AWSDatabaseMigrationServiceClientBuilder
.standard()
.withCredentials(awsCredentialsProvider)
.withRegion(regions)
.build();
}
}
}

View File

@ -0,0 +1,53 @@
/*
* 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.authentication.aws;
import java.util.Map;
import lombok.experimental.UtilityClass;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduce;
import com.amazonaws.services.elasticmapreduce.AmazonElasticMapReduceClientBuilder;
@UtilityClass
public class AmazonElasticMapReduceClientFactory {
public AmazonElasticMapReduce createAmazonElasticMapReduceClient(Map<String, String> awsProperties) {
AWSCredentialsProvider awsCredentialsProvider = AWSCredentialsProviderFactor.credentialsProvider(awsProperties);
Regions regions = Regions.fromName(awsProperties.get(AwsConfigurationKeys.AWS_REGION));
String endpoint = awsProperties.get(AwsConfigurationKeys.AWS_ENDPOINT);
if (endpoint != null && !endpoint.isEmpty()) {
return AmazonElasticMapReduceClientBuilder
.standard()
.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint, regions.getName()))
.withCredentials(awsCredentialsProvider)
.build();
} else {
return AmazonElasticMapReduceClientBuilder
.standard()
.withCredentials(awsCredentialsProvider)
.withRegion(regions)
.build();
}
}
}

View File

@ -0,0 +1,54 @@
/*
* 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.authentication.aws;
import java.util.Map;
import lombok.experimental.UtilityClass;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
@UtilityClass
public class AmazonS3ClientFactory {
public AmazonS3 createAmazonS3Client(Map<String, String> awsProperties) {
AWSCredentialsProvider awsCredentialsProvider = AWSCredentialsProviderFactor.credentialsProvider(awsProperties);
Regions regions = Regions.fromName(awsProperties.get(AwsConfigurationKeys.AWS_REGION));
String endpoint = awsProperties.get(AwsConfigurationKeys.AWS_ENDPOINT);
if (endpoint != null && !endpoint.isEmpty()) {
return AmazonS3ClientBuilder
.standard()
.withPathStyleAccessEnabled(true)
.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint, regions.getName()))
.withCredentials(awsCredentialsProvider)
.build();
} else {
return AmazonS3ClientBuilder
.standard()
.withCredentials(awsCredentialsProvider)
.withRegion(regions)
.build();
}
}
}

View File

@ -0,0 +1,53 @@
/*
* 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.authentication.aws;
import java.util.Map;
import lombok.experimental.UtilityClass;
import com.amazonaws.auth.AWSCredentialsProvider;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.sagemaker.AmazonSageMaker;
import com.amazonaws.services.sagemaker.AmazonSageMakerClientBuilder;
@UtilityClass
public class AmazonSageMakerClientFactory {
public AmazonSageMaker createAmazonSageMakerClient(Map<String, String> awsProperties) {
AWSCredentialsProvider awsCredentialsProvider = AWSCredentialsProviderFactor.credentialsProvider(awsProperties);
Regions regions = Regions.fromName(awsProperties.get(AwsConfigurationKeys.AWS_REGION));
String endpoint = awsProperties.get(AwsConfigurationKeys.AWS_ENDPOINT);
if (endpoint != null && !endpoint.isEmpty()) {
return AmazonSageMakerClientBuilder
.standard()
.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(endpoint, regions.getName()))
.withCredentials(awsCredentialsProvider)
.build();
} else {
return AmazonSageMakerClientBuilder
.standard()
.withCredentials(awsCredentialsProvider)
.withRegion(regions)
.build();
}
}
}

View File

@ -0,0 +1,28 @@
/*
* 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.authentication.aws;
public class AwsConfigurationKeys {
public static final String AWS_AUTHENTICATION_TYPE = "credentials.provider.type";
public static final String AWS_REGION = "region";
public static final String AWS_ENDPOINT = "endpoint";
public static final String AWS_ACCESS_KEY_ID = "access.key.id";
public static final String AWS_SECRET = "access.key.secret";
}

View File

@ -0,0 +1,47 @@
/*
* 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.authentication.aws;
import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
import software.amazon.awssdk.regions.Region;
import software.amazon.awssdk.services.datasync.DataSyncClient;
import java.util.Map;
import lombok.experimental.UtilityClass;
@UtilityClass
public class DataSyncClientFactory {
public DataSyncClient createDataSyncClient(Map<String, String> awsProperties) {
// todo: upgrade the version of aws sdk
String awsAccessKeyId = awsProperties.get(AwsConfigurationKeys.AWS_ACCESS_KEY_ID);
String awsSecretAccessKey = awsProperties.get(AwsConfigurationKeys.AWS_SECRET);
final AwsBasicCredentials basicAWSCredentials = AwsBasicCredentials.create(awsAccessKeyId, awsSecretAccessKey);
final AwsCredentialsProvider awsCredentialsProvider = StaticCredentialsProvider.create(basicAWSCredentials);
// create a datasync client
return DataSyncClient.builder()
.region(Region.of(awsProperties.get(AwsConfigurationKeys.AWS_REGION)))
.credentialsProvider(awsCredentialsProvider)
.build();
}
}

View File

@ -0,0 +1,65 @@
#
# 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.
#
aws:
s3:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: accessKey123
access.key.secret: secretKey123
region: us-east-1
bucket.name: dolphinscheduler
endpoint: http://s3:9000
emr:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: minioadmin
access.key.secret: minioadmin
region: cn-north-1
endpoint: http://localhost:9000
sagemaker:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: minioadmin
access.key.secret: minioadmin
region: cn-north-1
endpoint: http://localhost:9000
dms:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: minioadmin
access.key.secret: minioadmin
region: cn-north-1
endpoint: http://localhost:9000
datasync:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: minioadmin
access.key.secret: minioadmin
region: cn-north-1
endpoint: http://localhost:9000

View File

@ -0,0 +1,46 @@
<?xml version="1.0" encoding="UTF-8"?>
<!--
~ Licensed to the Apache Software Foundation (ASF) under one or more
~ contributor license agreements. See the NOTICE file distributed with
~ this work for additional information regarding copyright ownership.
~ The ASF licenses this file to You under the Apache License, Version 2.0
~ (the "License"); you may not use this file except in compliance with
~ the License. You may obtain a copy of the License at
~
~ http://www.apache.org/licenses/LICENSE-2.0
~
~ Unless required by applicable law or agreed to in writing, software
~ distributed under the License is distributed on an "AS IS" BASIS,
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
~ See the License for the specific language governing permissions and
~ limitations under the License.
-->
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler</artifactId>
<version>dev-SNAPSHOT</version>
</parent>
<artifactId>dolphinscheduler-authentication</artifactId>
<packaging>pom</packaging>
<modules>
<module>dolphinscheduler-aws-authentication</module>
</modules>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-bom</artifactId>
<version>${project.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
</project>

View File

@ -41,6 +41,10 @@
</dependencyManagement>
<dependencies>
<!-- dolphinscheduler -->
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-aws-authentication</artifactId>
</dependency>
<dependency>
<groupId>commons-io</groupId>
<artifactId>commons-io</artifactId>

View File

@ -36,6 +36,7 @@ public final class Constants {
public static final String COMMON_PROPERTIES_PATH = "/common.properties";
public static final String REMOTE_LOGGING_YAML_PATH = "/remote-logging.yaml";
public static final String AWS_YAML_PATH = "/aws.yaml";
public static final String FORMAT_SS = "%s%s";
public static final String FORMAT_S_S = "%s/%s";
@ -130,8 +131,7 @@ public final class Constants {
*/
public static final String RESOURCE_STORAGE_TYPE = "resource.storage.type";
public static final String AWS_S3_BUCKET_NAME = "resource.aws.s3.bucket.name";
public static final String AWS_END_POINT = "resource.aws.s3.endpoint";
public static final String AWS_S3_BUCKET_NAME = "aws.s3.bucket.name";
public static final String ALIBABA_CLOUD_OSS_BUCKET_NAME = "resource.alibaba.cloud.oss.bucket.name";
public static final String ALIBABA_CLOUD_OSS_END_POINT = "resource.alibaba.cloud.oss.endpoint";
@ -249,11 +249,6 @@ public final class Constants {
*/
public static final Pattern REGEX_USER_NAME = Pattern.compile("^[a-zA-Z0-9._-]{3,39}$");
/**
* file path regex
*/
public static final Pattern REGEX_FILE_PATH = Pattern.compile("^[a-zA-Z0-9_./-]+$");
/**
* read permission
*/
@ -704,19 +699,8 @@ public final class Constants {
public static final String REMOTE_LOGGING_OSS_ENDPOINT = "remote.logging.oss.endpoint";
/**
* remote logging for S3
*/
public static final String REMOTE_LOGGING_S3_ACCESS_KEY_ID = "remote.logging.s3.access.key.id";
public static final String REMOTE_LOGGING_S3_ACCESS_KEY_SECRET = "remote.logging.s3.access.key.secret";
public static final String REMOTE_LOGGING_S3_BUCKET_NAME = "remote.logging.s3.bucket.name";
public static final String REMOTE_LOGGING_S3_ENDPOINT = "remote.logging.s3.endpoint";
public static final String REMOTE_LOGGING_S3_REGION = "remote.logging.s3.region";
/**
* remote logging for GCS
*/

View File

@ -21,5 +21,5 @@ public class TenantConstants {
public static final String DEFAULT_TENANT_CODE = "default";
public static final String BOOTSTRAPT_SYSTEM_USER = System.getProperty("user.name");
public static final String BOOTSTRAP_SYSTEM_USER = System.getProperty("user.name");
}

View File

@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.common.log.remote;
import org.apache.dolphinscheduler.authentication.aws.AmazonS3ClientFactory;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
@ -26,41 +27,25 @@ import java.io.Closeable;
import java.io.File;
import java.io.FileOutputStream;
import java.io.IOException;
import java.util.Map;
import lombok.extern.slf4j.Slf4j;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.regions.Regions;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.S3ObjectInputStream;
@Slf4j
public class S3RemoteLogHandler implements RemoteLogHandler, Closeable {
private String accessKeyId;
private final String bucketName;
private String accessKeySecret;
private String region;
private String bucketName;
private String endPoint;
private AmazonS3 s3Client;
private final AmazonS3 s3Client;
private static S3RemoteLogHandler instance;
private S3RemoteLogHandler() {
accessKeyId = readAccessKeyID();
accessKeySecret = readAccessKeySecret();
region = readRegion();
bucketName = readBucketName();
endPoint = readEndPoint();
s3Client = buildS3Client();
checkBucketNameExists(bucketName);
}
@ -74,23 +59,8 @@ public class S3RemoteLogHandler implements RemoteLogHandler, Closeable {
}
protected AmazonS3 buildS3Client() {
if (StringUtils.isNotEmpty(endPoint)) {
return AmazonS3ClientBuilder
.standard()
.withPathStyleAccessEnabled(true)
.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(
endPoint, Regions.fromName(region).getName()))
.withCredentials(
new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKeyId, accessKeySecret)))
.build();
} else {
return AmazonS3ClientBuilder
.standard()
.withCredentials(
new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKeyId, accessKeySecret)))
.withRegion(Regions.fromName(region))
.build();
}
Map<String, String> awsProperties = PropertyUtils.getByPrefix("aws.s3.", "");
return AmazonS3ClientFactory.createAmazonS3Client(awsProperties);
}
@Override
@ -131,24 +101,8 @@ public class S3RemoteLogHandler implements RemoteLogHandler, Closeable {
}
}
protected String readAccessKeyID() {
return PropertyUtils.getString(Constants.REMOTE_LOGGING_S3_ACCESS_KEY_ID);
}
protected String readAccessKeySecret() {
return PropertyUtils.getString(Constants.REMOTE_LOGGING_S3_ACCESS_KEY_SECRET);
}
protected String readRegion() {
return PropertyUtils.getString(Constants.REMOTE_LOGGING_S3_REGION);
}
protected String readBucketName() {
return PropertyUtils.getString(Constants.REMOTE_LOGGING_S3_BUCKET_NAME);
}
protected String readEndPoint() {
return PropertyUtils.getString(Constants.REMOTE_LOGGING_S3_ENDPOINT);
return PropertyUtils.getString(Constants.AWS_S3_BUCKET_NAME);
}
public void checkBucketNameExists(String bucketName) {

View File

@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.common.utils;
import static org.apache.dolphinscheduler.common.constants.Constants.AWS_YAML_PATH;
import static org.apache.dolphinscheduler.common.constants.Constants.COMMON_PROPERTIES_PATH;
import static org.apache.dolphinscheduler.common.constants.Constants.REMOTE_LOGGING_YAML_PATH;
@ -42,7 +43,7 @@ public class PropertyUtils {
private final ImmutablePriorityPropertyDelegate propertyDelegate =
new ImmutablePriorityPropertyDelegate(
new ImmutablePropertyDelegate(COMMON_PROPERTIES_PATH),
new ImmutableYamlDelegate(REMOTE_LOGGING_YAML_PATH));
new ImmutableYamlDelegate(REMOTE_LOGGING_YAML_PATH, AWS_YAML_PATH));
public static String getString(String key) {
return propertyDelegate.get(key.trim());
@ -106,6 +107,19 @@ public class PropertyUtils {
return matchedProperties;
}
/**
* Get all properties with specified prefix, like: fs., will replace the prefix with newPrefix
*/
public static Map<String, String> getByPrefix(String prefix, String newPrefix) {
Map<String, String> matchedProperties = new HashMap<>();
for (String propName : propertyDelegate.getPropertyKeys()) {
if (propName.startsWith(prefix)) {
matchedProperties.put(propName.replace(prefix, newPrefix), propertyDelegate.get(propName));
}
}
return matchedProperties;
}
public static <T> Set<T> getSet(String key, Function<String, Set<T>> transformFunction, Set<T> defaultValue) {
return propertyDelegate.get(key, transformFunction, defaultValue);
}

View File

@ -39,17 +39,6 @@ resource.azure.tenant.id=minioadmin
# The query interval
resource.query.interval=10000
# The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id=minioadmin
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=minioadmin
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=cn-north-1
# The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name.
resource.aws.s3.bucket.name=dolphinscheduler
# You need to set this parameter when private cloud s3. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn
resource.aws.s3.endpoint=http://localhost:9000
# alibaba cloud access key id, required if you set resource.storage.type=OSS
resource.alibaba.cloud.access.key.id=<your-access-key-id>
# alibaba cloud access key secret, required if you set resource.storage.type=OSS
@ -188,16 +177,7 @@ remote.logging.oss.access.key.secret=<access.key.secret>
remote.logging.oss.bucket.name=<bucket.name>
# oss endpoint, required if you set remote.logging.target=OSS
remote.logging.oss.endpoint=<endpoint>
# s3 access key id, required if you set remote.logging.target=S3
remote.logging.s3.access.key.id=<access.key.id>
# s3 access key secret, required if you set remote.logging.target=S3
remote.logging.s3.access.key.secret=<access.key.secret>
# s3 bucket name, required if you set remote.logging.target=S3
remote.logging.s3.bucket.name=<bucket.name>
# s3 endpoint, required if you set remote.logging.target=S3
remote.logging.s3.endpoint=<endpoint>
# s3 region, required if you set remote.logging.target=S3
remote.logging.s3.region=<region>
# the location of the google cloud credential, required if you set remote.logging.target=GCS
remote.logging.google.cloud.storage.credential=/path/to/credential
# gcs bucket name, required if you set remote.logging.target=GCS

View File

@ -34,28 +34,16 @@ remote-logging:
bucket.name: <bucket.name>
# oss endpoint, required if you set remote-logging.target=OSS
endpoint: <endpoint>
# required if you set remote-logging.target=S3
s3:
# s3 access key id, required if you set remote-logging.target=S3
access.key.id: <access.key.id>
# s3 access key secret, required if you set remote-logging.target=S3
access.key.secret: <access.key.secret>
# s3 bucket name, required if you set remote-logging.target=S3
bucket.name: <bucket.name>
# s3 endpoint, required if you set remote-logging.target=S3
endpoint: <endpoint>
# s3 region, required if you set remote-logging.target=S3
region: <region>
google.cloud.storage:
# the location of the google cloud credential, required if you set remote-logging.target=GCS
credential: /path/to/credential
# gcs bucket name, required if you set remote-logging.target=GCS
# gcs bucket name, required if you set remote-logging.target=GCS
bucket.name: <your-bucket>
abs:
# abs account name, required if you set resource.storage.type=ABS
account.name: <your-account-name>
# abs account key, required if you set resource.storage.type=ABS
# abs account key, required if you set resource.storage.type=ABS
account.key: <your-account-key>
# abs container name, required if you set resource.storage.type=ABS
# abs container name, required if you set resource.storage.type=ABS
container.name: <your-container-name>

View File

@ -17,12 +17,15 @@
package org.apache.dolphinscheduler.common.utils;
import static com.google.common.truth.Truth.assertThat;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.commons.lang3.StringUtils;
import java.util.Arrays;
import java.util.Collections;
import java.util.Map;
import java.util.Set;
import java.util.stream.Collectors;
@ -48,4 +51,14 @@ public class PropertyUtilsTest {
}, Sets.newHashSet("docker0"));
Assertions.assertEquals(Sets.newHashSet("docker0"), networkInterface);
}
@Test
void getByPrefix() {
Map<String, String> awsProperties = PropertyUtils.getByPrefix("resource.aws.", "");
assertThat(awsProperties).containsEntry("access.key.id", "minioadmin");
assertThat(awsProperties).containsEntry("secret.access.key", "minioadmin");
assertThat(awsProperties).containsEntry("region", "cn-north-1");
assertThat(awsProperties).containsEntry("s3.bucket.name", "dolphinscheduler");
assertThat(awsProperties).containsEntry("endpoint", "http://localhost:9000");
}
}

View File

@ -45,6 +45,10 @@ resource.azure.tenant.id=minioadmin
# The query interval
resource.query.interval=10000
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
aws.credentials.provider.type=AWSStaticCredentialsProvider
# The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id=minioadmin
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
@ -54,7 +58,7 @@ resource.aws.region=cn-north-1
# The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name.
resource.aws.s3.bucket.name=dolphinscheduler
# You need to set this parameter when private cloud s3. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn
resource.aws.s3.endpoint=http://localhost:9000
resource.aws.endpoint=http://localhost:9000
# alibaba cloud access key id, required if you set resource.storage.type=OSS
resource.alibaba.cloud.access.key.id=<your-access-key-id>

View File

@ -35,6 +35,7 @@ import java.sql.DriverManager;
import java.sql.SQLException;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import com.alibaba.druid.sql.parser.SQLParserUtils;
import com.google.auto.service.AutoService;
@ -128,8 +129,10 @@ public class SQLServerDataSourceProcessor extends AbstractDataSourceProcessor {
@Override
public List<String> splitAndRemoveComment(String sql) {
String cleanSQL = SQLParserUtils.removeComment(sql, com.alibaba.druid.DbType.sqlserver);
return SQLParserUtils.split(cleanSQL, com.alibaba.druid.DbType.sqlserver);
return SQLParserUtils.splitAndRemoveComment(sql, com.alibaba.druid.DbType.sqlserver)
.stream()
.map(subSql -> subSql.concat(";"))
.collect(Collectors.toList());
}
private String transformOther(Map<String, String> otherMap) {

View File

@ -17,12 +17,15 @@
package org.apache.dolphinscheduler.plugin.datasource.sqlserver.param;
import static com.google.common.truth.Truth.assertThat;
import org.apache.dolphinscheduler.common.constants.DataSourceConstants;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.plugin.datasource.api.utils.PasswordUtils;
import org.apache.dolphinscheduler.spi.enums.DbType;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.jupiter.api.Assertions;
@ -95,4 +98,32 @@ public class SQLServerDataSourceProcessorTest {
Assertions.assertEquals(DataSourceConstants.SQLSERVER_VALIDATION_QUERY,
sqlServerDatasourceProcessor.getValidationQuery());
}
@Test
void splitAndRemoveComment_singleSelect() {
String sql = "select * from table;";
List<String> subSqls = sqlServerDatasourceProcessor.splitAndRemoveComment(sql);
assertThat(subSqls).hasSize(1);
assertThat(subSqls.get(0)).isEqualTo("select * from table;");
}
@Test
void splitAndRemoveComment_singleMerge() {
String sql = "MERGE\n" +
" [ TOP ( expression ) [ PERCENT ] ]\n" +
" [ INTO ] <target_table> [ WITH ( <merge_hint> ) ] [ [ AS ] table_alias ]\n" +
" USING <table_source> [ [ AS ] table_alias ]\n" +
" ON <merge_search_condition>\n" +
" [ WHEN MATCHED [ AND <clause_search_condition> ]\n" +
" THEN <merge_matched> ] [ ...n ]\n" +
" [ WHEN NOT MATCHED [ BY TARGET ] [ AND <clause_search_condition> ]\n" +
" THEN <merge_not_matched> ]\n" +
" [ WHEN NOT MATCHED BY SOURCE [ AND <clause_search_condition> ]\n" +
" THEN <merge_matched> ] [ ...n ]\n" +
" [ <output_clause> ]\n" +
" [ OPTION ( <query_hint> [ ,...n ] ) ];";
List<String> subSqls = sqlServerDatasourceProcessor.splitAndRemoveComment(sql);
assertThat(subSqls).hasSize(1);
assertThat(subSqls.get(0)).isEqualTo(sql);
}
}

View File

@ -525,7 +525,6 @@ The text of each license is also included at licenses/LICENSE-[project].txt.
reactor-netty-core 1.0.22: https://mvnrepository.com/artifact/io.projectreactor.netty/reactor-netty-core/1.0.22, Apache 2.0
reactor-netty-http 1.0.22: https://mvnrepository.com/artifact/io.projectreactor.netty/reactor-netty-http/1.0.22, Apache 2.0
DmJdbcDriver18 8.1.2.79: https://mvnrepository.com/artifact/com.dameng/DmJdbcDriver18/8.1.2.79, Apache 2.0
nimbus-jose-jwt 9.22: https://mvnrepository.com/artifact/com.nimbusds/nimbus-jose-jwt/9.22, Apache 2.0
woodstox-core 6.4.0: https://mvnrepository.com/artifact/com.fasterxml.woodstox/woodstox-core/6.4.0, Apache 2.0
auto-value 1.10.1: https://mvnrepository.com/artifact/com.google.auto.value/auto-value/1.10.1, Apache 2.0
auto-value-annotations 1.10.4: https://mvnrepository.com/artifact/com.google.auto.value/auto-value-annotations/1.10.4, Apache 2.0

View File

@ -0,0 +1,65 @@
#
# 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.
#
aws:
s3:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: accessKey123
access.key.secret: secretKey123
region: us-east-1
bucket.name: dolphinscheduler
endpoint: http://s3:9000
emr:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: minioadmin
access.key.secret: minioadmin
region: cn-north-1
endpoint: http://localhost:9000
sagemaker:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: minioadmin
access.key.secret: minioadmin
region: cn-north-1
endpoint: http://localhost:9000
dms:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: minioadmin
access.key.secret: minioadmin
region: cn-north-1
endpoint: http://localhost:9000
datasync:
# The AWS credentials provider type. support: AWSStaticCredentialsProvider, InstanceProfileCredentialsProvider
# AWSStaticCredentialsProvider: use the access key and secret key to authenticate
# InstanceProfileCredentialsProvider: use the IAM role to authenticate
credentials.provider.type: AWSStaticCredentialsProvider
access.key.id: minioadmin
access.key.secret: minioadmin
region: cn-north-1
endpoint: http://localhost:9000

View File

@ -37,17 +37,6 @@ resource.azure.tenant.id=minioadmin
# The query interval
resource.query.interval=10000
# The AWS access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.access.key.id=accessKey123
# The AWS secret access key. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.secret.access.key=secretKey123
# The AWS Region to use. if resource.storage.type=S3 or use EMR-Task, This configuration is required
resource.aws.region=us-east-1
# The name of the bucket. You need to create them by yourself. Otherwise, the system cannot start. All buckets in Amazon S3 share a single namespace; ensure the bucket is given a unique name.
resource.aws.s3.bucket.name=dolphinscheduler
# You need to set this parameter when private cloud s3. If S3 uses public cloud, you only need to set resource.aws.region or set to the endpoint of a public cloud such as S3.cn-north-1.amazonaws.com.cn
resource.aws.s3.endpoint=http://s3:9000
# alibaba cloud access key id, required if you set resource.storage.type=OSS
resource.alibaba.cloud.access.key.id=<your-access-key-id>
# alibaba cloud access key secret, required if you set resource.storage.type=OSS

View File

@ -34,6 +34,7 @@ services:
retries: 120
volumes:
- ./common.properties:/opt/dolphinscheduler/conf/common.properties
- ./aws.yaml:/opt/dolphinscheduler/conf/aws.yaml
depends_on:
s3:
condition: service_healthy

View File

@ -141,7 +141,7 @@ public class NettyRemotingClient implements AutoCloseable {
return iRpcResponse;
}
private Channel getOrCreateChannel(Host host) {
Channel getOrCreateChannel(Host host) {
Channel channel = channels.get(host);
if (channel != null && channel.isActive()) {
return channel;
@ -166,13 +166,10 @@ public class NettyRemotingClient implements AutoCloseable {
* @param host host
* @return channel
*/
private Channel createChannel(Host host) {
Channel createChannel(Host host) {
try {
ChannelFuture future;
synchronized (bootstrap) {
future = bootstrap.connect(new InetSocketAddress(host.getIp(), host.getPort()));
}
future.await(clientConfig.getConnectTimeoutMillis());
ChannelFuture future = bootstrap.connect(new InetSocketAddress(host.getIp(), host.getPort()));
future = future.sync();
if (future.isSuccess()) {
return future.channel();
} else {

View File

@ -56,6 +56,13 @@
</includes>
<outputDirectory>conf</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-authentication/dolphinscheduler-aws-authentication/src/main/resources</directory>
<includes>
<include>**/*.yaml</include>
</includes>
<outputDirectory>conf</outputDirectory>
</fileSet>
</fileSets>
<dependencySets>
<dependencySet>

View File

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

View File

@ -66,7 +66,6 @@ public class TaskExecutionContextBuilder {
taskExecutionContext.setWorkerGroup(taskInstance.getWorkerGroup());
taskExecutionContext.setEnvironmentConfig(taskInstance.getEnvironmentConfig());
taskExecutionContext.setHost(taskInstance.getHost());
taskExecutionContext.setDelayTime(taskInstance.getDelayTime());
taskExecutionContext.setVarPool(taskInstance.getVarPool());
taskExecutionContext.setDryRun(taskInstance.getDryRun());
taskExecutionContext.setTestFlag(taskInstance.getTestFlag());

View File

@ -28,10 +28,9 @@ public interface StateEventHandler {
* @param stateEvent given state event.
* @throws StateEventHandleException this exception means it can be recovered.
* @throws StateEventHandleError this exception means it cannot be recovered, so the event need to drop.
* @throws StateEventHandleException this means it can be recovered.
*/
boolean handleStateEvent(WorkflowExecuteRunnable workflowExecuteRunnable,
StateEvent stateEvent) throws StateEventHandleException, StateEventHandleError, StateEventHandleFailure;
StateEvent stateEvent) throws StateEventHandleException, StateEventHandleError;
StateEventType getEventType();
}

View File

@ -52,4 +52,32 @@ public abstract class BaseTaskExecuteRunnable implements TaskExecuteRunnable {
return taskExecutionContext;
}
@Override
public int compareTo(TaskExecuteRunnable other) {
if (other == null) {
return 1;
}
int workflowInstancePriorityCompareResult = workflowInstance.getProcessInstancePriority().getCode() -
other.getWorkflowInstance().getProcessInstancePriority().getCode();
if (workflowInstancePriorityCompareResult != 0) {
return workflowInstancePriorityCompareResult;
}
// smaller number, higher priority
int taskInstancePriorityCompareResult = taskInstance.getTaskInstancePriority().getCode()
- other.getTaskInstance().getTaskInstancePriority().getCode();
if (taskInstancePriorityCompareResult != 0) {
return taskInstancePriorityCompareResult;
}
// larger number, higher priority
int taskGroupPriorityCompareResult =
taskInstance.getTaskGroupPriority() - other.getTaskInstance().getTaskGroupPriority();
if (taskGroupPriorityCompareResult != 0) {
return -taskGroupPriorityCompareResult;
}
// earlier submit time, higher priority
return taskInstance.getFirstSubmitTime().compareTo(other.getTaskInstance().getFirstSubmitTime());
}
}

View File

@ -24,7 +24,7 @@ import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.server.master.runner.operator.TaskExecuteRunnableOperatorManager;
public class DefaultTaskExecuteRunnable extends PriorityDelayTaskExecuteRunnable {
public class DefaultTaskExecuteRunnable extends BaseTaskExecuteRunnable {
private final TaskExecuteRunnableOperatorManager taskExecuteRunnableOperatorManager;

View File

@ -17,7 +17,8 @@
package org.apache.dolphinscheduler.server.master.runner;
import java.util.concurrent.DelayQueue;
import org.apache.dolphinscheduler.server.master.runner.queue.DelayEntry;
import org.apache.dolphinscheduler.server.master.runner.queue.PriorityDelayQueue;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
@ -25,26 +26,42 @@ import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Component;
/**
* The class is used to store {@link TaskExecuteRunnable} which needs to be dispatched. The {@link TaskExecuteRunnable} will be stored in a {@link DelayQueue},
* if the {@link TaskExecuteRunnable}'s delay time is 0, then it will be consumed by {@link GlobalTaskDispatchWaitingQueueLooper}.
* The class is used to store {@link TaskExecuteRunnable} which needs to be dispatched. The {@link TaskExecuteRunnable}
* will be stored in {@link PriorityDelayQueue}, if the {@link TaskExecuteRunnable}'s delay time is 0, then it will be
* consumed by {@link GlobalTaskDispatchWaitingQueueLooper}.
* <p>
* The order of {@link TaskExecuteRunnable} in the {@link PriorityDelayQueue} is determined by {@link TaskExecuteRunnable#compareTo}.
*/
@Slf4j
@Component
public class GlobalTaskDispatchWaitingQueue {
private final DelayQueue<DefaultTaskExecuteRunnable> queue = new DelayQueue<>();
private final PriorityDelayQueue<DelayEntry<TaskExecuteRunnable>> priorityDelayQueue = new PriorityDelayQueue<>();
public void submitTaskExecuteRunnable(DefaultTaskExecuteRunnable priorityTaskExecuteRunnable) {
queue.put(priorityTaskExecuteRunnable);
/**
* Submit a {@link TaskExecuteRunnable} with delay time 0, it will be consumed immediately.
*/
public void dispatchTaskExecuteRunnable(TaskExecuteRunnable taskExecuteRunnable) {
dispatchTaskExecuteRunnableWithDelay(taskExecuteRunnable, 0);
}
/**
* Submit a {@link TaskExecuteRunnable} with delay time, if the delay time <= 0 then it can be consumed.
*/
public void dispatchTaskExecuteRunnableWithDelay(TaskExecuteRunnable taskExecuteRunnable, long delayTimeMills) {
priorityDelayQueue.add(new DelayEntry<>(delayTimeMills, taskExecuteRunnable));
}
/**
* Consume {@link TaskExecuteRunnable} from the {@link PriorityDelayQueue}, only the delay time <= 0 can be consumed.
*/
@SneakyThrows
public DefaultTaskExecuteRunnable takeTaskExecuteRunnable() {
return queue.take();
public TaskExecuteRunnable takeTaskExecuteRunnable() {
return priorityDelayQueue.take().getData();
}
public int getWaitingDispatchTaskNumber() {
return queue.size();
return priorityDelayQueue.size();
}
}

View File

@ -18,13 +18,11 @@
package org.apache.dolphinscheduler.server.master.runner;
import org.apache.dolphinscheduler.common.thread.BaseDaemonThread;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.server.master.runner.dispatcher.TaskDispatchFactory;
import org.apache.dolphinscheduler.server.master.runner.dispatcher.TaskDispatcher;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import lombok.extern.slf4j.Slf4j;
@ -43,10 +41,6 @@ public class GlobalTaskDispatchWaitingQueueLooper extends BaseDaemonThread imple
private final AtomicBoolean RUNNING_FLAG = new AtomicBoolean(false);
private final AtomicInteger DISPATCHED_CONSECUTIVE_FAILURE_TIMES = new AtomicInteger();
private static final Integer MAX_DISPATCHED_FAILED_TIMES = 100;
public GlobalTaskDispatchWaitingQueueLooper() {
super("GlobalTaskDispatchWaitingQueueLooper");
}
@ -64,29 +58,34 @@ public class GlobalTaskDispatchWaitingQueueLooper extends BaseDaemonThread imple
@Override
public void run() {
DefaultTaskExecuteRunnable defaultTaskExecuteRunnable;
while (RUNNING_FLAG.get()) {
defaultTaskExecuteRunnable = globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable();
try {
TaskExecutionStatus status = defaultTaskExecuteRunnable.getTaskInstance().getState();
if (status != TaskExecutionStatus.SUBMITTED_SUCCESS && status != TaskExecutionStatus.DELAY_EXECUTION) {
log.warn("The TaskInstance {} state is : {}, will not dispatch",
defaultTaskExecuteRunnable.getTaskInstance().getName(), status);
continue;
}
doDispatch();
}
}
TaskDispatcher taskDispatcher =
taskDispatchFactory.getTaskDispatcher(defaultTaskExecuteRunnable.getTaskInstance());
taskDispatcher.dispatchTask(defaultTaskExecuteRunnable);
DISPATCHED_CONSECUTIVE_FAILURE_TIMES.set(0);
} catch (Exception e) {
defaultTaskExecuteRunnable.getTaskExecutionContext().increaseDispatchFailTimes();
globalTaskDispatchWaitingQueue.submitTaskExecuteRunnable(defaultTaskExecuteRunnable);
if (DISPATCHED_CONSECUTIVE_FAILURE_TIMES.incrementAndGet() > MAX_DISPATCHED_FAILED_TIMES) {
ThreadUtils.sleep(10 * 1000L);
}
log.error("Dispatch Task: {} failed", defaultTaskExecuteRunnable.getTaskInstance().getName(), e);
void doDispatch() {
final TaskExecuteRunnable taskExecuteRunnable = globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable();
TaskInstance taskInstance = taskExecuteRunnable.getTaskInstance();
if (taskInstance == null) {
// This case shouldn't happen, but if it does, log an error and continue
log.error("The TaskInstance is null, drop it(This case shouldn't happen)");
return;
}
try {
TaskExecutionStatus status = taskInstance.getState();
if (status != TaskExecutionStatus.SUBMITTED_SUCCESS && status != TaskExecutionStatus.DELAY_EXECUTION) {
log.warn("The TaskInstance {} state is : {}, will not dispatch", taskInstance.getName(), status);
return;
}
taskDispatchFactory.getTaskDispatcher(taskInstance).dispatchTask(taskExecuteRunnable);
} catch (Exception e) {
// If dispatch failed, will put the task back to the queue
// The task will be dispatched after waiting time.
// the waiting time will increase multiple of times, but will not exceed 60 seconds
long waitingTimeMills = Math.max(
taskExecuteRunnable.getTaskExecutionContext().increaseDispatchFailTimes() * 1_000L, 60_000L);
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnableWithDelay(taskExecuteRunnable, waitingTimeMills);
log.error("Dispatch Task: {} failed will retry after: {}/ms", taskInstance.getName(), waitingTimeMills, e);
}
}

View File

@ -1,85 +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.runner;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
public abstract class PriorityDelayTaskExecuteRunnable extends BaseTaskExecuteRunnable implements Delayed {
public PriorityDelayTaskExecuteRunnable(ProcessInstance workflowInstance,
TaskInstance taskInstance,
TaskExecutionContext taskExecutionContext) {
super(workflowInstance, taskInstance, taskExecutionContext);
}
@Override
public long getDelay(TimeUnit unit) {
return unit.convert(
DateUtils.getRemainTime(taskExecutionContext.getFirstSubmitTime(),
taskExecutionContext.getDelayTime() * 60L),
TimeUnit.SECONDS);
}
@Override
public int compareTo(Delayed o) {
if (o == null) {
return 1;
}
int delayTimeCompareResult =
Long.compare(this.getDelay(TimeUnit.MILLISECONDS), o.getDelay(TimeUnit.MILLISECONDS));
if (delayTimeCompareResult != 0) {
return delayTimeCompareResult;
}
PriorityDelayTaskExecuteRunnable other = (PriorityDelayTaskExecuteRunnable) o;
// the smaller dispatch fail times, the higher priority
int dispatchFailTimesCompareResult = taskExecutionContext.getDispatchFailTimes()
- other.getTaskExecutionContext().getDispatchFailTimes();
if (dispatchFailTimesCompareResult != 0) {
return dispatchFailTimesCompareResult;
}
int workflowInstancePriorityCompareResult = workflowInstance.getProcessInstancePriority().getCode()
- other.getWorkflowInstance().getProcessInstancePriority().getCode();
if (workflowInstancePriorityCompareResult != 0) {
return workflowInstancePriorityCompareResult;
}
long workflowInstanceIdCompareResult = workflowInstance.getId().compareTo(other.getWorkflowInstance().getId());
if (workflowInstanceIdCompareResult != 0) {
return workflowInstancePriorityCompareResult;
}
int taskInstancePriorityCompareResult = taskInstance.getTaskInstancePriority().getCode()
- other.getTaskInstance().getTaskInstancePriority().getCode();
if (taskInstancePriorityCompareResult != 0) {
return taskInstancePriorityCompareResult;
}
// larger number, higher priority
int taskGroupPriorityCompareResult =
taskInstance.getTaskGroupPriority() - other.getTaskInstance().getTaskGroupPriority();
if (taskGroupPriorityCompareResult != 0) {
return -taskGroupPriorityCompareResult;
}
// The task instance shouldn't be equals
return taskInstance.getId().compareTo(other.getTaskInstance().getId());
}
}

View File

@ -25,7 +25,7 @@ import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
* This interface is used to define a task which is executing.
* todo: split to MasterTaskExecuteRunnable and WorkerTaskExecuteRunnable
*/
public interface TaskExecuteRunnable {
public interface TaskExecuteRunnable extends Comparable<TaskExecuteRunnable> {
void dispatch();
@ -40,4 +40,5 @@ public interface TaskExecuteRunnable {
TaskInstance getTaskInstance();
TaskExecutionContext getTaskExecutionContext();
}

View File

@ -65,7 +65,6 @@ import org.apache.dolphinscheduler.server.master.config.MasterConfig;
import org.apache.dolphinscheduler.server.master.event.StateEvent;
import org.apache.dolphinscheduler.server.master.event.StateEventHandleError;
import org.apache.dolphinscheduler.server.master.event.StateEventHandleException;
import org.apache.dolphinscheduler.server.master.event.StateEventHandleFailure;
import org.apache.dolphinscheduler.server.master.event.StateEventHandler;
import org.apache.dolphinscheduler.server.master.event.StateEventHandlerManager;
import org.apache.dolphinscheduler.server.master.event.TaskStateEvent;
@ -298,13 +297,6 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
stateEvent,
stateEventHandleException);
ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
} catch (StateEventHandleFailure stateEventHandleFailure) {
log.error("State event handle failed, will move event to the tail: {}",
stateEvent,
stateEventHandleFailure);
this.stateEvents.remove(stateEvent);
this.stateEvents.offer(stateEvent);
ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS);
} catch (Exception e) {
// we catch the exception here, since if the state event handle failed, the state event will still
// keep
@ -647,6 +639,7 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
command.setProcessInstanceId(0);
command.setProcessDefinitionVersion(workflowInstance.getProcessDefinitionVersion());
command.setTestFlag(workflowInstance.getTestFlag());
command.setTenantCode(workflowInstance.getTenantCode());
int create = commandService.createCommand(command);
processService.saveCommandTrigger(command.getId(), workflowInstance.getId());
return create;
@ -1226,12 +1219,12 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
|| state == TaskExecutionStatus.DISPATCH
|| state == TaskExecutionStatus.SUBMITTED_SUCCESS
|| state == TaskExecutionStatus.DELAY_EXECUTION) {
// try to take over task instance
if (state == TaskExecutionStatus.SUBMITTED_SUCCESS || state == TaskExecutionStatus.DELAY_EXECUTION
|| state == TaskExecutionStatus.DISPATCH) {
if (state == TaskExecutionStatus.SUBMITTED_SUCCESS
|| state == TaskExecutionStatus.DELAY_EXECUTION) {
// The taskInstance is not in running, directly takeover it
} else if (tryToTakeOverTaskInstance(existTaskInstance)) {
log.info("Success take over task {}", existTaskInstance.getName());
// If the taskInstance has already dispatched to worker then will try to take-over it
log.info("Success take over task {} -> status: {}", existTaskInstance.getName(), state);
continue;
} else {
// set the task instance state to fault tolerance
@ -1366,7 +1359,7 @@ public class WorkflowExecuteRunnable implements IWorkflowExecuteRunnable {
}
}
log.info("The dependTasks of task all success, currentTaskCode: {}, dependTaskCodes: {}",
taskCode, Arrays.toString(completeTaskSet.toArray()));
taskCode, Arrays.toString(indirectDepCodeList.toArray()));
return DependResult.SUCCESS;
}

View File

@ -17,14 +17,13 @@
package org.apache.dolphinscheduler.server.master.runner.operator;
import org.apache.dolphinscheduler.common.utils.DateUtils;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.dao.repository.TaskInstanceDao;
import org.apache.dolphinscheduler.plugin.task.api.enums.TaskExecutionStatus;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.GlobalTaskDispatchWaitingQueue;
import java.util.concurrent.TimeUnit;
import lombok.extern.slf4j.Slf4j;
@Slf4j
@ -43,16 +42,17 @@ public abstract class BaseTaskExecuteRunnableDispatchOperator implements TaskExe
@Override
public void operate(DefaultTaskExecuteRunnable taskExecuteRunnable) {
long remainTime = taskExecuteRunnable.getDelay(TimeUnit.SECONDS);
TaskInstance taskInstance = taskExecuteRunnable.getTaskInstance();
if (remainTime > 0) {
long remainTimeMills =
DateUtils.getRemainTime(taskInstance.getFirstSubmitTime(), taskInstance.getDelayTime() * 60L) * 1_000;
if (remainTimeMills > 0) {
taskInstance.setState(TaskExecutionStatus.DELAY_EXECUTION);
taskInstanceDao.updateById(taskInstance);
log.info("Current taskInstance: {} is choose delay execution, delay time: {}/min, remainTime: {}/s",
log.info("Current taskInstance: {} is choose delay execution, delay time: {}/min, remainTime: {}/ms",
taskInstance.getName(),
taskInstance.getDelayTime(),
remainTime);
remainTimeMills);
}
globalTaskDispatchWaitingQueue.submitTaskExecuteRunnable(taskExecuteRunnable);
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnableWithDelay(taskExecuteRunnable, remainTimeMills);
}
}

View File

@ -0,0 +1,82 @@
/*
* 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.runner.queue;
import static com.google.common.base.Preconditions.checkNotNull;
import java.util.Objects;
import java.util.concurrent.Delayed;
import java.util.concurrent.TimeUnit;
import lombok.Getter;
import org.jetbrains.annotations.NotNull;
public class DelayEntry<V extends Comparable<V>> implements Delayed {
private final long delayTimeMills;
private final long triggerTimeMills;
@Getter
private final V data;
public DelayEntry(long delayTimeMills, V data) {
this.delayTimeMills = delayTimeMills;
this.triggerTimeMills = System.currentTimeMillis() + delayTimeMills;
this.data = checkNotNull(data, "data is null");
}
@Override
public long getDelay(@NotNull TimeUnit unit) {
long remainTimeMills = triggerTimeMills - System.currentTimeMillis();
if (TimeUnit.MILLISECONDS.equals(unit)) {
return remainTimeMills;
}
return unit.convert(remainTimeMills, TimeUnit.MILLISECONDS);
}
@Override
public int compareTo(@NotNull Delayed o) {
DelayEntry<V> other = (DelayEntry<V>) o;
int delayTimeMillsCompareResult = Long.compare(delayTimeMills, other.delayTimeMills);
if (delayTimeMillsCompareResult != 0) {
return delayTimeMillsCompareResult;
}
if (data == null || other.data == null) {
return 0;
}
return data.compareTo(other.data);
}
@Override
public boolean equals(Object o) {
if (this == o)
return true;
if (o == null || getClass() != o.getClass())
return false;
DelayEntry<?> that = (DelayEntry<?>) o;
return delayTimeMills == that.delayTimeMills && Objects.equals(data, that.data);
}
@Override
public int hashCode() {
return Objects.hash(delayTimeMills, data);
}
}

View File

@ -15,19 +15,27 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.master.event;
package org.apache.dolphinscheduler.server.master.runner.queue;
/**
* This exception represent the exception can be recovered, when we get this exception,
* we will move the event to the fail of the queue.
*/
public class StateEventHandleFailure extends Exception {
import java.util.concurrent.DelayQueue;
public StateEventHandleFailure(String message) {
super(message);
import lombok.SneakyThrows;
public class PriorityDelayQueue<V extends DelayEntry> {
private final DelayQueue<V> queue = new DelayQueue<>();
public void add(V v) {
queue.put(v);
}
public StateEventHandleFailure(String message, Throwable throwable) {
super(message, throwable);
@SneakyThrows
public V take() {
return queue.take();
}
public int size() {
return queue.size();
}
}

View File

@ -61,26 +61,30 @@ public class SwitchLogicTask extends BaseSyncLogicTask<SwitchParameters> {
this.taskInstance = workflowExecuteRunnable.getTaskInstance(taskExecutionContext.getTaskInstanceId())
.orElseThrow(() -> new LogicTaskInitializeException(
"Cannot find the task instance in workflow execute runnable"));
// Since the default branch is not in the dependTaskList, we need to add it to the end
// otherwise the default branch will never be skipped in DAGHelper
addDefaultBranchToEnd();
}
@Override
public void handle() throws MasterTaskExecuteException {
// Calculate the condition result and get the next node
if (CollectionUtils.isEmpty(taskParameters.getDependTaskList())) {
// If the branch is empty then will go into the default branch
// This case shouldn't happen, we can directly throw exception and forbid the user to set branch
moveToDefaultBranch();
} else {
calculateSwitchBranch();
}
taskInstance.setSwitchDependency(taskParameters);
log.info("Switch task execute finished");
taskExecutionContext.setCurrentExecutionStatus(TaskExecutionStatus.SUCCESS);
log.info("Switch task execute finished: {}", taskExecutionContext.getCurrentExecutionStatus().name());
}
private void moveToDefaultBranch() {
checkIfBranchExist(taskParameters.getNextNode());
List<SwitchResultVo> switchResultVos = taskParameters.getDependTaskList();
switchResultVos.add(new SwitchResultVo(null, taskParameters.getNextNode()));
SwitchResultVo defaultSwitchResultVo = getDefaultSwitchResultVo();
checkIfBranchExist(defaultSwitchResultVo.getNextNode());
taskParameters.setResultConditionLocation(switchResultVos.size() - 1);
log.info("The condition is not satisfied, move to the default branch: {}",
@ -90,9 +94,6 @@ public class SwitchLogicTask extends BaseSyncLogicTask<SwitchParameters> {
private void calculateSwitchBranch() {
List<SwitchResultVo> switchResultVos = taskParameters.getDependTaskList();
if (CollectionUtils.isEmpty(switchResultVos)) {
moveToDefaultBranch();
}
Map<String, Property> globalParams = taskExecutionContext.getPrepareParamsMap();
Map<String, Property> varParams = JSONUtils
.toList(taskInstance.getVarPool(), Property.class)
@ -100,7 +101,8 @@ public class SwitchLogicTask extends BaseSyncLogicTask<SwitchParameters> {
.collect(Collectors.toMap(Property::getProp, Property -> Property));
int finalConditionLocation = -1;
for (int i = 0; i < switchResultVos.size(); i++) {
// The last one is the default branch, no need to calculate
for (int i = 0; i < switchResultVos.size() - 1; i++) {
SwitchResultVo switchResultVo = switchResultVos.get(i);
log.info("Begin to execute {} condition: {} ", i, switchResultVo.getCondition());
String content = SwitchTaskUtils.generateContentWithTaskParams(switchResultVo.getCondition(), globalParams,
@ -111,14 +113,18 @@ public class SwitchLogicTask extends BaseSyncLogicTask<SwitchParameters> {
result = SwitchTaskUtils.evaluate(content);
log.info("Execute condition sentence: {} successfully: {}", content, result);
if (result) {
// If matched, break the loop
finalConditionLocation = i;
break;
}
} catch (Exception e) {
log.info("Execute condition sentence: {} failed", content, e);
}
}
// If the finalConditionLocation is -1, then the default branch will be executed
if (finalConditionLocation >= 0) {
checkIfBranchExist(switchResultVos.get(finalConditionLocation).getNextNode());
List<Long> nextNodes = switchResultVos.get(finalConditionLocation).getNextNode();
checkIfBranchExist(nextNodes);
log.info("The condition is satisfied, move to the branch: {}",
switchResultVos.get(finalConditionLocation).getNextNode().stream()
.map(node -> workflowExecuteRunnable.getWorkflowExecuteContext().getWorkflowGraph().getDag()
@ -126,7 +132,6 @@ public class SwitchLogicTask extends BaseSyncLogicTask<SwitchParameters> {
.collect(Collectors.toList()));
taskParameters.setResultConditionLocation(finalConditionLocation);
} else {
log.info("All conditions are not satisfied, move to the default branch");
moveToDefaultBranch();
}
}
@ -147,4 +152,16 @@ public class SwitchLogicTask extends BaseSyncLogicTask<SwitchParameters> {
}
}
private void addDefaultBranchToEnd() {
SwitchResultVo switchResultVo = new SwitchResultVo(null, taskParameters.getNextNode());
List<SwitchResultVo> dependTaskList = taskParameters.getDependTaskList();
if (!dependTaskList.contains(switchResultVo)) {
dependTaskList.add(switchResultVo);
}
}
private SwitchResultVo getDefaultSwitchResultVo() {
return taskParameters.getDependTaskList().get(taskParameters.getDependTaskList().size() - 1);
}
}

View File

@ -63,7 +63,6 @@ public class SwitchTaskUtils {
if (MapUtils.isNotEmpty(varParams)) {
params.putAll(varParams);
}
String originContent = content;
Pattern pattern = Pattern.compile(rgex);
Matcher m = pattern.matcher(content);
while (m.find()) {
@ -82,10 +81,6 @@ public class SwitchTaskUtils {
content = content.replace("${" + paramName + "}", value);
}
// if not replace any params, throw exception to avoid illegal condition
if (originContent.equals(content)) {
throw new IllegalArgumentException("condition is not valid, please check it. condition: " + condition);
}
return content;
}

View File

@ -0,0 +1,184 @@
/*
* 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.runner;
import static com.google.common.truth.Truth.assertThat;
import static org.awaitility.Awaitility.await;
import static org.junit.jupiter.api.Assertions.assertThrowsExactly;
import org.apache.dolphinscheduler.common.enums.Priority;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.server.master.runner.operator.TaskExecuteRunnableOperatorManager;
import org.apache.commons.lang3.time.DateUtils;
import java.time.Duration;
import java.util.Date;
import java.util.concurrent.CompletableFuture;
import org.awaitility.Awaitility;
import org.awaitility.core.ConditionTimeoutException;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.BeforeEach;
import org.junit.jupiter.api.Test;
class GlobalTaskDispatchWaitingQueueTest {
private GlobalTaskDispatchWaitingQueue globalTaskDispatchWaitingQueue;
@BeforeEach
public void setUp() {
globalTaskDispatchWaitingQueue = new GlobalTaskDispatchWaitingQueue();
}
@Test
void submitTaskExecuteRunnable() {
TaskExecuteRunnable taskExecuteRunnable = createTaskExecuteRunnable();
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable);
Awaitility.await()
.atMost(Duration.ofSeconds(1))
.untilAsserted(
() -> Assertions.assertNotNull(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable()));
}
@Test
void testSubmitTaskExecuteRunnableWithDelay() {
TaskExecuteRunnable taskExecuteRunnable = createTaskExecuteRunnable();
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnableWithDelay(taskExecuteRunnable, 3_000L);
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable);
assertThat(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable()).isNotNull();
Awaitility.await()
.atLeast(Duration.ofSeconds(2))
.atMost(Duration.ofSeconds(4))
.untilAsserted(
() -> Assertions.assertNotNull(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable()));
}
@Test
void takeTaskExecuteRunnable_NoElementShouldBlock() {
CompletableFuture<Void> completableFuture =
CompletableFuture.runAsync(() -> globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable());
assertThrowsExactly(ConditionTimeoutException.class,
() -> await()
.atLeast(Duration.ofSeconds(2))
.timeout(Duration.ofSeconds(3))
.until(completableFuture::isDone));
}
@Test
void takeTaskExecuteRunnable_withDifferentTaskInstancePriority() {
TaskExecuteRunnable taskExecuteRunnable1 = createTaskExecuteRunnable();
taskExecuteRunnable1.getTaskInstance().setId(1);
taskExecuteRunnable1.getTaskInstance().setTaskInstancePriority(Priority.MEDIUM);
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable1);
TaskExecuteRunnable taskExecuteRunnable2 = createTaskExecuteRunnable();
taskExecuteRunnable2.getTaskInstance().setId(2);
taskExecuteRunnable2.getTaskInstance().setTaskInstancePriority(Priority.HIGH);
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable2);
TaskExecuteRunnable taskExecuteRunnable3 = createTaskExecuteRunnable();
taskExecuteRunnable3.getTaskInstance().setId(3);
taskExecuteRunnable3.getTaskInstance().setTaskInstancePriority(Priority.LOW);
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable3);
assertThat(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable().getTaskInstance().getId())
.isEqualTo(2);
assertThat(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable().getTaskInstance().getId())
.isEqualTo(1);
assertThat(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable().getTaskInstance().getId())
.isEqualTo(3);
}
@Test
void takeTaskExecuteRunnable_withDifferentTaskGroupPriority() {
TaskExecuteRunnable taskExecuteRunnable1 = createTaskExecuteRunnable();
taskExecuteRunnable1.getTaskInstance().setId(1);
taskExecuteRunnable1.getTaskInstance().setTaskGroupPriority(Priority.MEDIUM.getCode());
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable1);
TaskExecuteRunnable taskExecuteRunnable2 = createTaskExecuteRunnable();
taskExecuteRunnable2.getTaskInstance().setId(2);
taskExecuteRunnable2.getTaskInstance().setTaskGroupPriority(Priority.HIGH.getCode());
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable2);
TaskExecuteRunnable taskExecuteRunnable3 = createTaskExecuteRunnable();
taskExecuteRunnable3.getTaskInstance().setId(3);
taskExecuteRunnable3.getTaskInstance().setTaskGroupPriority(Priority.LOW.getCode());
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable3);
assertThat(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable().getTaskInstance().getId())
.isEqualTo(3);
assertThat(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable().getTaskInstance().getId())
.isEqualTo(1);
assertThat(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable().getTaskInstance().getId())
.isEqualTo(2);
}
@Test
void takeTaskExecuteRunnable_withDifferentSubmitTime() {
Date now = new Date();
TaskExecuteRunnable taskExecuteRunnable1 = createTaskExecuteRunnable();
taskExecuteRunnable1.getTaskInstance().setId(1);
taskExecuteRunnable1.getTaskInstance().setFirstSubmitTime(now);
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable1);
TaskExecuteRunnable taskExecuteRunnable2 = createTaskExecuteRunnable();
taskExecuteRunnable2.getTaskInstance().setId(2);
taskExecuteRunnable2.getTaskInstance().setFirstSubmitTime(DateUtils.addMinutes(now, 1));
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable2);
TaskExecuteRunnable taskExecuteRunnable3 = createTaskExecuteRunnable();
taskExecuteRunnable3.getTaskInstance().setId(3);
taskExecuteRunnable3.getTaskInstance().setFirstSubmitTime(DateUtils.addMinutes(now, -1));
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable3);
assertThat(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable().getTaskInstance().getId())
.isEqualTo(3);
assertThat(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable().getTaskInstance().getId())
.isEqualTo(1);
assertThat(globalTaskDispatchWaitingQueue.takeTaskExecuteRunnable().getTaskInstance().getId())
.isEqualTo(2);
}
@Test
void getWaitingDispatchTaskNumber() {
Assertions.assertEquals(0, globalTaskDispatchWaitingQueue.getWaitingDispatchTaskNumber());
TaskExecuteRunnable taskExecuteRunnable = createTaskExecuteRunnable();
globalTaskDispatchWaitingQueue.dispatchTaskExecuteRunnable(taskExecuteRunnable);
Assertions.assertEquals(1, globalTaskDispatchWaitingQueue.getWaitingDispatchTaskNumber());
}
private TaskExecuteRunnable createTaskExecuteRunnable() {
ProcessInstance processInstance = new ProcessInstance();
processInstance.setProcessInstancePriority(Priority.MEDIUM);
TaskInstance taskInstance = new TaskInstance();
taskInstance.setTaskInstancePriority(Priority.MEDIUM);
taskInstance.setFirstSubmitTime(new Date());
TaskExecutionContext taskExecutionContext = new TaskExecutionContext();
return new DefaultTaskExecuteRunnable(processInstance, taskInstance, taskExecutionContext,
new TaskExecuteRunnableOperatorManager());
}
}

View File

@ -1,67 +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.runner.execute;
import org.apache.dolphinscheduler.common.enums.Priority;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
import org.apache.dolphinscheduler.dao.entity.TaskInstance;
import org.apache.dolphinscheduler.plugin.task.api.TaskExecutionContext;
import org.apache.dolphinscheduler.server.master.runner.DefaultTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.PriorityDelayTaskExecuteRunnable;
import org.apache.dolphinscheduler.server.master.runner.operator.TaskExecuteRunnableOperatorManager;
import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
public class PriorityDelayTaskExecuteRunnableTest {
@Test
public void testCompareTo() {
TaskExecuteRunnableOperatorManager taskOperatorManager = new TaskExecuteRunnableOperatorManager();
ProcessInstance workflowInstance = new ProcessInstance();
workflowInstance.setId(1);
workflowInstance.setProcessInstancePriority(Priority.HIGH);
TaskInstance t1 = new TaskInstance();
t1.setId(1);
t1.setTaskInstancePriority(Priority.HIGH);
TaskInstance t2 = new TaskInstance();
t2.setId(1);
t2.setTaskInstancePriority(Priority.HIGH);
TaskExecutionContext context1 = new TaskExecutionContext();
TaskExecutionContext context2 = new TaskExecutionContext();
PriorityDelayTaskExecuteRunnable p1 =
new DefaultTaskExecuteRunnable(workflowInstance, t1, context1, taskOperatorManager);
PriorityDelayTaskExecuteRunnable p2 =
new DefaultTaskExecuteRunnable(workflowInstance, t2, context2, taskOperatorManager);
Assertions.assertEquals(0, p1.compareTo(p2));
// the higher priority, the higher priority
t2.setTaskInstancePriority(Priority.MEDIUM);
Assertions.assertTrue(p1.compareTo(p2) < 0);
// the smaller dispatch fail times, the higher priority
context1.setDispatchFailTimes(1);
Assertions.assertTrue(p1.compareTo(p2) > 0);
}
}

View File

@ -0,0 +1,35 @@
/*
* 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.runner.queue;
import java.util.concurrent.TimeUnit;
import org.junit.jupiter.api.Test;
import com.google.common.truth.Truth;
class DelayEntryTest {
@Test
void getDelay() {
DelayEntry<String> delayEntry = new DelayEntry<>(1_000L, "Item");
Truth.assertThat(delayEntry.getDelay(TimeUnit.NANOSECONDS))
.isWithin(100)
.of(TimeUnit.NANOSECONDS.convert(1_000L, TimeUnit.MILLISECONDS));
}
}

View File

@ -50,7 +50,7 @@ public class SwitchTaskUtilsTest {
Map<String, Property> globalParams = new HashMap<>();
Map<String, Property> varParams = new HashMap<>();
globalParams.put("test", new Property("test", Direct.IN, DataType.INTEGER, "1"));
Assertions.assertThrowsExactly(IllegalArgumentException.class, () -> {
Assertions.assertDoesNotThrow(() -> {
SwitchTaskUtils.generateContentWithTaskParams(content, globalParams, varParams);
});
@ -70,15 +70,5 @@ public class SwitchTaskUtilsTest {
SwitchTaskUtils.evaluate(script);
});
String contentWithSpecify1 = "cmd.abc";
Assertions.assertThrowsExactly(IllegalArgumentException.class, () -> {
SwitchTaskUtils.generateContentWithTaskParams(contentWithSpecify1, globalParams, varParams);
});
String contentWithSpecify2 = "cmd()";
Assertions.assertThrowsExactly(IllegalArgumentException.class, () -> {
SwitchTaskUtils.generateContentWithTaskParams(contentWithSpecify2, globalParams, varParams);
});
}
}

View File

@ -36,16 +36,10 @@ import com.google.common.base.Preconditions;
*/
public class StandByTaskInstancePriorityQueue implements TaskPriorityQueue<TaskInstance> {
/**
* queue size
*/
private static final Integer QUEUE_MAX_SIZE = 3000;
/**
* queue
*/
private final PriorityQueue<TaskInstance> queue =
new PriorityQueue<>(QUEUE_MAX_SIZE, new TaskInstancePriorityComparator());
private final PriorityQueue<TaskInstance> queue = new PriorityQueue<>(new TaskInstancePriorityComparator());
private final Set<String> taskInstanceIdentifySet = Collections.synchronizedSet(new HashSet<>());
/**

View File

@ -86,6 +86,13 @@
</includes>
<outputDirectory>conf</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-authentication/dolphinscheduler-aws-authentication/src/main/resources</directory>
<includes>
<include>**/*.yaml</include>
</includes>
<outputDirectory>conf</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-api/src/main/resources</directory>
<includes>

View File

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

View File

@ -17,19 +17,18 @@
package org.apache.dolphinscheduler.plugin.storage.s3;
import static org.apache.dolphinscheduler.common.constants.Constants.AWS_END_POINT;
import static org.apache.dolphinscheduler.common.constants.Constants.FOLDER_SEPARATOR;
import static org.apache.dolphinscheduler.common.constants.Constants.FORMAT_S_S;
import static org.apache.dolphinscheduler.common.constants.Constants.RESOURCE_TYPE_FILE;
import static org.apache.dolphinscheduler.common.constants.Constants.RESOURCE_TYPE_UDF;
import org.apache.dolphinscheduler.authentication.aws.AmazonS3ClientFactory;
import org.apache.dolphinscheduler.common.constants.Constants;
import org.apache.dolphinscheduler.common.enums.ResUploadType;
import org.apache.dolphinscheduler.common.utils.FileUtils;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.plugin.storage.api.StorageEntity;
import org.apache.dolphinscheduler.plugin.storage.api.StorageOperate;
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
import org.apache.commons.lang3.StringUtils;
@ -57,11 +56,7 @@ import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import com.amazonaws.AmazonServiceException;
import com.amazonaws.auth.AWSStaticCredentialsProvider;
import com.amazonaws.auth.BasicAWSCredentials;
import com.amazonaws.client.builder.AwsClientBuilder;
import com.amazonaws.services.s3.AmazonS3;
import com.amazonaws.services.s3.AmazonS3ClientBuilder;
import com.amazonaws.services.s3.model.AmazonS3Exception;
import com.amazonaws.services.s3.model.DeleteObjectsRequest;
import com.amazonaws.services.s3.model.ListObjectsV2Request;
@ -79,71 +74,27 @@ import com.amazonaws.services.s3.transfer.TransferManagerBuilder;
@Data
public class S3StorageOperator implements Closeable, StorageOperate {
private String accessKeyId;
private String accessKeySecret;
private String region;
private String bucketName;
private String endPoint;
private AmazonS3 s3Client;
public S3StorageOperator() {
}
public void init() {
accessKeyId = readAccessKeyID();
accessKeySecret = readAccessKeySecret();
region = readRegion();
bucketName = readBucketName();
endPoint = readEndPoint();
s3Client = buildS3Client();
checkBucketNameExists(bucketName);
}
protected AmazonS3 buildS3Client() {
if (!StringUtils.isEmpty(endPoint)) {
return AmazonS3ClientBuilder
.standard()
.withPathStyleAccessEnabled(true)
.withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(
endPoint, region))
.withCredentials(
new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKeyId, accessKeySecret)))
.build();
} else {
return AmazonS3ClientBuilder
.standard()
.withCredentials(
new AWSStaticCredentialsProvider(new BasicAWSCredentials(accessKeyId, accessKeySecret)))
.withRegion(region)
.build();
}
}
protected String readAccessKeyID() {
return PropertyUtils.getString(TaskConstants.AWS_ACCESS_KEY_ID);
}
protected String readAccessKeySecret() {
return PropertyUtils.getString(TaskConstants.AWS_SECRET_ACCESS_KEY);
}
protected String readRegion() {
return PropertyUtils.getString(TaskConstants.AWS_REGION);
return AmazonS3ClientFactory.createAmazonS3Client(PropertyUtils.getByPrefix("aws.s3.", ""));
}
protected String readBucketName() {
return PropertyUtils.getString(Constants.AWS_S3_BUCKET_NAME);
}
protected String readEndPoint() {
return PropertyUtils.getString(AWS_END_POINT);
}
@Override
public void close() throws IOException {
s3Client.shutdown();

View File

@ -78,13 +78,7 @@ public class S3StorageOperatorTest {
public void setUp() throws Exception {
s3StorageOperator = Mockito.spy(new S3StorageOperator());
doReturn(ACCESS_KEY_ID_MOCK).when(s3StorageOperator)
.readAccessKeyID();
doReturn(ACCESS_KEY_SECRET_MOCK).when(s3StorageOperator)
.readAccessKeySecret();
doReturn(REGION_MOCK).when(s3StorageOperator).readRegion();
doReturn(BUCKET_NAME_MOCK).when(s3StorageOperator).readBucketName();
doReturn(END_POINT_MOCK).when(s3StorageOperator).readEndPoint();
Mockito.doReturn(s3Client)
.when(s3StorageOperator).buildS3Client();
Mockito.doNothing()
@ -96,9 +90,6 @@ public class S3StorageOperatorTest {
@Test
public void testInit() {
verify(s3StorageOperator, times(1)).buildS3Client();
Assertions.assertEquals(ACCESS_KEY_ID_MOCK, s3StorageOperator.getAccessKeyId());
Assertions.assertEquals(ACCESS_KEY_SECRET_MOCK, s3StorageOperator.getAccessKeySecret());
Assertions.assertEquals(REGION_MOCK, s3StorageOperator.getRegion());
Assertions.assertEquals(BUCKET_NAME_MOCK, s3StorageOperator.getBucketName());
}

View File

@ -203,11 +203,6 @@ public class TaskExecutionContext implements Serializable {
*/
private String workerGroup;
/**
* delay execution time.
*/
private int delayTime;
/**
* current execution status
*/
@ -262,12 +257,9 @@ public class TaskExecutionContext implements Serializable {
private boolean logBufferEnable;
/**
* dispatch fail times
*/
private int dispatchFailTimes;
public void increaseDispatchFailTimes() {
this.dispatchFailTimes++;
public int increaseDispatchFailTimes() {
return ++dispatchFailTimes;
}
}

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