[cherry-pick-3.0.0-alpha] Cherry pick dev to 3.0.0 alpha (#9429)

* [Fix][UI Next][V1.0.0-Alpha]Add zh for dag execution policy (#9363)

* [Bug-9235][Alert]Fix wechat markdown message and change wechat form structure (#9367)

* fix wechat issues:
1. change table msg type to markdown.
2. change userId to not required and enrich hints
3. change 'app id' to 'app id and chat id'

* fix wechat issues:
1. revert table showtype and add markdown showtype.
2. enrich hints.
3. delete 'chatid', rename agentid to weChatAgentIdChatId.
4. modify code to send markdown message.

* fix wechat issues: Change the language pack of agentId to agentId/chatId.

* fix format

* fix param name

Co-authored-by: Amy <amywang0104@163.com>

* [FIX-9355] Fix scheduleTime of start-process-instance api in api-doc (#9359)

* fix #9355

* fix #9355

* fix ut error

* fix ut error

* [CI] try to fix ci (#9366)

* try to fix ci

* try to fix ci

* try to fix ci

* try to fix ci

* try to fix ci

* try to fix ci

* try to fix ci

* try to fix ci

* try to fix ci

* try to fix ci

* try to fix ci

* try to fix ci

* try to fix ci

* [optimization] [Service] Optimization ProcessService and add ProcessService interface (#9370)

* [task-spark][docs] Corrected notice section (#9375)

* [python] Migrate pythonGatewayServer into api server (#9372)

Currently the size of our distribute package is up to
800MB, this patch is migrate python gateway server into
api server

The distribute package size before and after this patch is:

```sh
# before
796M   apache-dolphinscheduler-2.0.4-SNAPSHOT-bin.tar.gz

# after
647M   apache-dolphinscheduler-2.0.4-SNAPSHOT-bin.tar.gz
```

* [Fix][UI Next][V1.0.0-Alpha] Add light color theme to echarts. (#9381)

* [Bug][API-9364]fix ProcessInstance wrong alert group id (#9383)

* fix ProcessInstance wrong alert group id

* change  createComplementCommandList method to protected

* [BUG][WORKER-9349]fix param priority (#9379)

* fix param priority

* fix params priority code logic

* [Improvement] change method access (#9390)

* change method to protected

* change method access

* [Fix-9221] [alert-server] optimization and gracefully close (#9246)

* [Fix-9221] [alert-server] optimization and gracefully close

This closes #9221

* [Fix-9221] [alert-server] remove unused mock data

This closes #9221

* [Fix-9221] [alert-server] remove unused mock data

This closes #9221

* [Fix-9221] [alert-server] remove unnecessary Mockito stubbings

* [Fix-9221] [alert-server] init AlertPluginManager in AlertServer

* [Fix-9221] [alert-server] AlertServerTest add AlertPluginManager installPlugin

* [Fix-9221] [alert-server] replace @Eventlistener with @PostConstruct

* [Fix-9221] [alert-server] sonar check solution

* [Improvement-9221] [alert] update constructor injection and replace IStoppable with Closeable

Co-authored-by: guoshupei <guoshupei@lixiang.com>

* [Fix][UI Next][V1.0.0-Alpha] Fix the task instance forced success button multi-language support error. (#9392)

* [doc] Change get help from dev mail list to slack (#9377)

* Change all get help from dev mailing list to slack, because
  we find out mailing list have many users ask for subscribe
  and they maybe subscribe by accident.
* remove join dev mailing list in faq.md because we already
  have it in https://dolphinscheduler.apache.org/en-us/community/development/subscribe.html

* Add new code owner of docs module (#9388)

* [CI] Enable CI to remove unexpected files in /docs/img dir (#9393)

* [Bug][UI Next]Modify the display state logic of save buttons under workflow definition (#9403)

* Modifying site Configurations

* Modify the display state logic of save buttons under workflow definition

* [doc] Remove observability (#9402)

SkyWalking v9 is coming soon and there are without
DolphinScheduler menus anymore, So we should remove
the SW agent to avoid confusion.

close: #9242

* [DS-9387][refactor]Remove the lock in the start method of the MasterRegistryClient class (#9389)

* [Fix-9251] [WORKER] reslove the sql task about of add the udf resource failed (#9319)

* feat(resource  manager): extend s3 to the storage of ds

1.fix some spell question
2.extend the type of storage
3.add the s3utils
to manager resource
4.automatic inject the storage in addition to your
config

* fix(resource  manager): update the dependency

* fix(resource  manager): extend s3 to the storage of ds

fix the constant of hadooputils

* fix(resource  manager): extend s3 to the storage of ds

1.fix some spell question
2.delete the import *

* fix(resource  manager):

merge  the unitTest:
1.TenantServiceImpl
2.ResourceServiceImpl
3.UserServiceImpl

* fix(resource  manager): extend s3 to the storage of ds

merge the resourceServiceTest

* fix(resource  manager): test  cancel the test method

createTenant verifyTenant

* fix(resource  manager): merge the code  follow the check-result of sonar

* fix(resource  manager): extend s3 to the storage of ds

fit the spell question

* fix(resource  manager): extend s3 to the storage of ds

revert the common.properties

* fix(resource  manager): extend s3 to the storage of ds

update the storageConfig with None

* fix(resource  manager): extend s3 to the storage of ds

fix the judge of resourceType

* fix(resource  manager): extend s3 to the storage of ds

undo the compile-mysql

* fix(resource  manager): extend s3 to the storage of ds

delete hadoop aws

* fix(resource  manager): extend s3 to the storage of ds

update the know-dependencies to delete aws 1.7.4
update the e2e
file-manager common.properties

* fix(resource  manager): extend s3 to the storage of ds

update the aws-region

* fix(resource  manager): extend s3 to the storage of ds

fix the storageconfig init

* fix(resource  manager): update e2e docker-compose

update e2e docker-compose

* fix(resource  manager): extend s3 to the storage of ds

revent the e2e common.proprites

print the resource type in propertyUtil

* fix(resource  manager): extend s3 to the storage of ds
1.println the properties

* fix(resource  manager): println the s3 info

* fix(resource  manager): extend s3 to the storage of ds

delete the info  and upgrade the s3 info to e2e

* fix(resource  manager): extend s3 to the storage of ds

add the bucket init

* fix(resource  manager): extend s3 to the storage of ds

1.fix some spell question
2.delete the import *

* fix(resource  manager): extend s3 to the storage of ds

upgrade the s3 endpoint

* fix(resource  manager): withPathStyleAccessEnabled(true)

* fix(resource  manager): extend s3 to the storage of ds

1.fix some spell question
2.delete the import *

* fix(resource  manager): upgrade the  s3client builder

* fix(resource  manager): correct  the s3 point to s3client

* fix(resource  manager): update the constant BUCKET_NAME

* fix(resource  manager): e2e  s3 endpoint -> s3:9000

* fix(resource  manager): extend s3 to the storage of ds

1.fix some spell question
2.delete the import *

* style(resource  manager): add info to createBucket

* style(resource  manager): debug the log

* ci(resource  manager): test

test s3

* ci(ci): add INSERT INTO dolphinscheduler.t_ds_tenant (id, tenant_code, description, queue_id, create_time, update_time) VALUES(1, 'root', NULL, 1, NULL, NULL); to h2.sql

* fix(resource  manager): update the h2 sql

* fix(resource  manager): solve to delete the tenant

* style(resource  manager): merge the style end delete the unuse s3 config

* fix(resource  manager): extend s3 to the storage of ds

UPDATE the rename resources when s3

* fix(resource  manager): extend s3 to the storage of ds

1.fix the code style of QuartzImpl

* fix(resource  manager): extend s3 to the storage of ds

1.impoort restore_type to CommonUtils

* fix(resource  manager): update the work thread

* fix(resource  manager): update  the baseTaskProcessor

* fix(resource  manager): upgrade dolphinscheduler-standalone-server.xml

* fix(resource  manager): add  user Info to dolphinscheduler_h2.sql

* fix(resource  manager): merge  the resourceType to NONE

* style(upgrade the log level to info):

* fix(resource  manager): sysnc the h2.sql

* fix(resource  manager): update the merge the user tenant

* fix(resource  manager): merge the resourcesServiceImpl

* fix(resource  manager):

when the storage is s3 ,that the directory can't be renamed

* fix(resource  manager): in s3 ,the directory cannot be renamed

* fix(resource  manager): delete the deleteRenameDirectory in E2E

* fix(resource  manager): check the style and  recoverd the test

* fix(resource  manager): delete the log.print(LoginUser)

* fix(server): fix the  udf serialize

* fix(master  task): update the udfTest to update the json string

* fix(test): update the udfFuncTest

* fix(common): syn the common.properties

* fix(udfTest): upgrade the udfTest

* fix(common): revent the common.properties

* [Fix-9316] [Task] Configure DB2 data source SQL script execution report ResultSet has been closed exception in SQL task  (#9317)

* fix db2 error in the sql task

* update limit in sql task

* [UI] Migrate NPM to PNPM in CI builds (#9431)

Co-authored-by: Devosend <devosend@gmail.com>
Co-authored-by: Tq <tianqitobethefirst@gmail.com>
Co-authored-by: Amy <amywang0104@163.com>
Co-authored-by: xiangzihao <460888207@qq.com>
Co-authored-by: gaojun2048 <gaojun2048@gmail.com>
Co-authored-by: mans2singh <mans2singh@users.noreply.github.com>
Co-authored-by: Jiajie Zhong <zhongjiajie955@hotmail.com>
Co-authored-by: Amy0104 <97265214+Amy0104@users.noreply.github.com>
Co-authored-by: guoshupei <15764973965@163.com>
Co-authored-by: guoshupei <guoshupei@lixiang.com>
Co-authored-by: songjianet <1778651752@qq.com>
Co-authored-by: Eric Gao <ericgao.apache@gmail.com>
Co-authored-by: labbomb <739955946@qq.com>
Co-authored-by: worry <7039986@qq.com>
Co-authored-by: nobolity <nobolity@users.noreply.github.com>
Co-authored-by: Kerwin <37063904+zhuangchong@users.noreply.github.com>
Co-authored-by: kezhenxu94 <kezhenxu94@apache.org>
This commit is contained in:
caishunfeng 2022-04-11 16:50:26 +08:00 committed by GitHub
parent ff6a3bd6dd
commit 9fda8c5811
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
87 changed files with 3723 additions and 3823 deletions

1
.github/CODEOWNERS vendored
View File

@ -20,3 +20,4 @@ dolphinscheduler/dolphinscheduler-e2e @kezhenxu94
dolphinscheduler/dolphinscheduler-registry @kezhenxu94
dolphinscheduler/dolphinscheduler-standalone-server @kezhenxu94
dolphinscheduler/dolphinscheduler-python @zhongjiajie
dolphinscheduler/docs @zhongjiajie @Tianqi-Dotes

View File

@ -22,11 +22,12 @@ labels: [ "bug", "Waiting for reply" ]
body:
- type: markdown
attributes:
value: |
value: >
Please make sure what you are reporting is indeed a bug with reproducible steps, if you want to ask questions
or share ideas, you can head to our
[Discussions](https://github.com/apache/dolphinscheduler/discussions) tab, you can also [subscribe to our mailing list](mailto:dev-subscribe@dolphinscheduler.apache.org) and send
emails to [our mailing list](mailto:dev@dolphinscheduler.apache.org)
[Discussions](https://github.com/apache/dolphinscheduler/discussions) tab, you can also
[join our slack](https://join.slack.com/t/asf-dolphinscheduler/shared_invite/zt-omtdhuio-_JISsxYhiVsltmC5h38yfw)
and send your question to channel `#troubleshooting`
For better global communication, Please write in English.

View File

@ -42,18 +42,18 @@ jobs:
name: Backend-Path-Filter
runs-on: ubuntu-latest
outputs:
ignore: ${{ steps.filter.outputs.ignore }}
not-ignore: ${{ steps.filter.outputs.not-ignore }}
steps:
- uses: dorny/paths-filter@b2feaf19c27470162a626bd6fa8438ae5b263721
id: filter
with:
filters: |
ignore:
- '(docs/**|dolphinscheduler-ui/**|dolphinscheduler-ui-next/**)'
not-ignore:
- '!(docs/**|dolphinscheduler-ui/**|dolphinscheduler-ui-next/**)'
build:
name: Backend-Build
needs: paths-filter
if: ${{ needs.paths-filter.outputs.ignore == 'false' }}
if: ${{ needs.paths-filter.outputs.not-ignore == 'true' }}
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
@ -81,13 +81,16 @@ jobs:
name: Build
runs-on: ubuntu-latest
timeout-minutes: 30
needs: [ build ]
needs: [ build, paths-filter ]
if: always()
steps:
- name: Status
run: |
if [[ ${{ needs.build.result }} == 'success' || ${{ needs.paths-filter.outputs.ignore == 'true' }} ]]; then
echo "Passed!"
else
if [[ ${{ needs.paths-filter.outputs.not-ignore }} == 'false' ]]; then
echo "Skip Build!"
exit 0
fi
if [[ ${{ needs.build.result }} != 'success' ]]; then
echo "Build Failed!"
exit -1
fi

View File

@ -33,18 +33,18 @@ jobs:
name: E2E-Path-Filter
runs-on: ubuntu-latest
outputs:
ignore: ${{ steps.filter.outputs.ignore }}
not-ignore: ${{ steps.filter.outputs.not-ignore }}
steps:
- uses: dorny/paths-filter@b2feaf19c27470162a626bd6fa8438ae5b263721
id: filter
with:
filters: |
ignore:
- '(docs/**)'
not-ignore:
- '!(docs/**)'
build:
name: E2E-Build
needs: paths-filter
if: ${{ needs.paths-filter.outputs.ignore == 'false' }}
if: ${{ needs.paths-filter.outputs.not-ignore == 'true' }}
runs-on: ubuntu-latest
timeout-minutes: 20
steps:
@ -155,13 +155,16 @@ jobs:
name: E2E
runs-on: ubuntu-latest
timeout-minutes: 30
needs: [ e2e ]
needs: [ e2e, paths-filter ]
if: always()
steps:
- name: Status
run: |
if [[ ${{ needs.e2e.result }} == 'success' || ${{ needs.paths-filter.outputs.ignore == 'true' }} ]]; then
echo "Passed!"
else
if [[ ${{ needs.paths-filter.outputs.not-ignore }} == 'false' ]]; then
echo "Skip E2E!"
exit 0
fi
if [[ ${{ needs.e2e.result }} != 'success' ]]; then
echo "E2E Failed!"
exit -1
fi

View File

@ -58,6 +58,7 @@ jobs:
node-version: 16
- name: Compile and Build
run: |
npm install
npm run lint
npm run build:prod
npm install pnpm -g
pnpm install
pnpm run lint
pnpm run build:prod

View File

@ -40,5 +40,8 @@ jobs:
- name: "Comment in issue"
uses: ./.github/actions/comment-on-issue
with:
message: "Hi:\n* Thank you for your feedback, we have received your issue, Please wait patiently for a reply.\n* In order for us to understand your request as soon as possible, please provide detailed information、version or pictures.\n* If you haven't received a reply for a long time, you can subscribe to the developer's emailMail subscription steps reference https://dolphinscheduler.apache.org/en-us/community/development/subscribe.html ,Then write the issue URL in the email content and send question to dev@dolphinscheduler.apache.org."
message: |
Thank you for your feedback, we have received your issue, Please wait patiently for a reply.
* In order for us to understand your request as soon as possible, please provide detailed information、version or pictures.
* If you haven't received a reply for a long time, you can [join our slack](https://join.slack.com/t/asf-dolphinscheduler/shared_invite/zt-omtdhuio-_JISsxYhiVsltmC5h38yfw) and send your question to channel `#troubleshooting`
GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}

View File

@ -40,18 +40,18 @@ jobs:
name: Unit-Test-Path-Filter
runs-on: ubuntu-latest
outputs:
ignore: ${{ steps.filter.outputs.ignore }}
not-ignore: ${{ steps.filter.outputs.not-ignore }}
steps:
- uses: dorny/paths-filter@b2feaf19c27470162a626bd6fa8438ae5b263721
id: filter
with:
filters: |
ignore:
- '(docs/**)'
not-ignore:
- '!(docs/**)'
unit-test:
name: Unit-Test
needs: paths-filter
if: ${{ needs.paths-filter.outputs.ignore == 'false' }}
if: ${{ needs.paths-filter.outputs.not-ignore == 'true' }}
runs-on: ubuntu-latest
timeout-minutes: 30
steps:
@ -114,13 +114,16 @@ jobs:
name: Unit Test
runs-on: ubuntu-latest
timeout-minutes: 30
needs: [ unit-test ]
needs: [ unit-test, paths-filter ]
if: always()
steps:
- name: Status
run: |
if [[ ${{ needs.unit-test.result }} == 'success' || ${{ needs.paths-filter.outputs.ignore == 'true' }} ]]; then
echo "Passed!"
else
if [[ ${{ needs.paths-filter.outputs.not-ignore }} == 'false' ]]; then
echo "Skip Unit Test!"
exit 0
fi
if [[ ${{ needs.unit-test.result }} != 'success' ]]; then
echo "Unit Test Failed!"
exit -1
fi

View File

@ -86,7 +86,7 @@ We would like to express our deep gratitude to all the open-source projects used
## Get Help
1. Submit an [issue](https://github.com/apache/dolphinscheduler/issues/new/choose)
1. Subscribe to this mailing list: https://dolphinscheduler.apache.org/en-us/community/development/subscribe.html, then email dev@dolphinscheduler.apache.org
2. [Join our slack](https://join.slack.com/t/asf-dolphinscheduler/shared_invite/zt-omtdhuio-_JISsxYhiVsltmC5h38yfw) and send your question to channel `#troubleshooting`
## Community

View File

@ -87,8 +87,8 @@ Dolphin Scheduler使用了很多优秀的开源项目比如google的guava、g
## 获得帮助
1. 提交issue
2. 先订阅邮件开发列表:[订阅邮件列表](https://dolphinscheduler.apache.org/zh-cn/community/development/subscribe.html), 订阅成功后发送邮件到dev@dolphinscheduler.apache.org.
1. 提交 [issue](https://github.com/apache/dolphinscheduler/issues/new/choose)
2. [加入slack群](https://join.slack.com/t/asf-dolphinscheduler/shared_invite/zt-omtdhuio-_JISsxYhiVsltmC5h38yfw) 并在频道 `#troubleshooting` 中提问
## 社区

View File

@ -140,29 +140,6 @@ services:
networks:
- dolphinscheduler
dolphinscheduler-python-gateway:
image: ${HUB}/dolphinscheduler-python:${TAG}
ports:
- "54321:54321"
- "25333:25333"
env_file: .env
healthcheck:
test: [ "CMD", "curl", "http://localhost:54321/actuator/health" ]
interval: 30s
timeout: 5s
retries: 3
depends_on:
dolphinscheduler-schema-initializer:
condition: service_completed_successfully
dolphinscheduler-zookeeper:
condition: service_healthy
volumes:
- dolphinscheduler-logs:/opt/dolphinscheduler/logs
- dolphinscheduler-shared-local:/opt/soft
- dolphinscheduler-resource-local:/dolphinscheduler
networks:
- dolphinscheduler
networks:
dolphinscheduler:
driver: bridge

View File

@ -118,27 +118,6 @@ services:
mode: replicated
replicas: 1
dolphinscheduler-python-gateway:
image: apache/dolphinscheduler-python-gateway
ports:
- 54321:54321
- 25333:25333
env_file: .env
healthcheck:
test: [ "CMD", "curl", "http://localhost:54321/actuator/health" ]
interval: 30s
timeout: 5s
retries: 3
volumes:
- dolphinscheduler-logs:/opt/dolphinscheduler/logs
- dolphinscheduler-shared-local:/opt/soft
- dolphinscheduler-resource-local:/dolphinscheduler
networks:
- dolphinscheduler
deploy:
mode: replicated
replicas: 1
networks:
dolphinscheduler:
driver: overlay

View File

@ -44,9 +44,6 @@ Create default docker images' fullname.
{{- define "dolphinscheduler.image.fullname.tools" -}}
{{- .Values.image.registry }}/dolphinscheduler-tools:{{ .Values.image.tag | default .Chart.AppVersion -}}
{{- end -}}
{{- define "dolphinscheduler.image.fullname.python-gateway" -}}
{{- .Values.image.registry }}/dolphinscheduler-python-gateway:{{ .Values.image.tag | default .Chart.AppVersion -}}
{{- end -}}
{{/*
Create a default common labels.

View File

@ -298,16 +298,6 @@ export default {
},
],
},
{
title: 'Observability',
children: [
{
title: 'SkyWalking-Agent',
link: '/en-us/docs/dev/user_doc/guide/installation/skywalking-agent.html',
},
],
},
{
title: 'FAQ',
children: [
@ -609,15 +599,6 @@ export default {
},
],
},
{
title: '可观测性',
children: [
{
title: 'SkyWalking-Agent',
link: '/zh-cn/docs/dev/user_doc/guide/installation/skywalking-agent.html',
},
],
},
{
title: 'FAQ',
children: [

View File

@ -544,17 +544,6 @@ A: 1, edit nginx config file /etc/nginx/conf.d/escheduler.conf
---
## Q : Welcome to subscribe the DolphinScheduler development mailing list
A: In the process of using DolphinScheduler, if you have any questions or ideas, suggestions, you can participate in the DolphinScheduler community building through the Apache mailing list. Sending a subscription email is also very simple, the steps are as follows:
1, Send an email to dev-subscribe@dolphinscheduler.apache.org with your own email address, subject and content.
2, Receive confirmation email and reply. After completing step 1, you will receive a confirmation email from dev-help@dolphinscheduler.apache.org (if not received, please confirm whether the email is automatically classified as spam, promotion email, subscription email, etc.) . Then reply directly to the email, or click on the link in the email to reply quickly, the subject and content are arbitrary.
3, Receive a welcome email. After completing the above steps, you will receive a welcome email with the subject WELCOME to dev@dolphinscheduler.apache.org, and you have successfully subscribed to the Apache DolphinScheduler mailing list.
---
## Q : Workflow Dependency
A: 1, It is currently judged according to natural days, at the end of last month: the judgment time is the workflow A start_time/scheduler_time between '2019-05-31 00:00:00' and '2019-05-31 23:59:59'. Last month: It is judged that there is an A instance completed every day from the 1st to the end of the month. Last week: There are completed A instances 7 days last week. The first two days: Judging yesterday and the day before yesterday, there must be a completed A instance for two days.
@ -712,6 +701,12 @@ AThe repair can be completed by executing the following SQL in the database:
update t_ds_version set version='2.0.1';
```
## Can not find python-gateway-server in distribute package
After version 3.0.0-alpha, Python gateway server integrate into API server, and Python gateway service will start when you
start API server. If you want disabled when Python gateway service you could change API server configuration in path
`api-server/conf/application.yaml` and change attribute `python-gateway.enabled : false`.
---
## We will collect more FAQ later

View File

@ -193,7 +193,9 @@ sh ./bin/dolphinscheduler-daemon.sh start alert-server
sh ./bin/dolphinscheduler-daemon.sh stop alert-server
```
> **_Note:_**: Please refer to the section of "System Architecture Design" for service usage
> **_Note:_**: Please refer to the section of "System Architecture Design" for service usage. Python gateway service is
> started along with the api-server, and if you do not want to start Python gateway service please disabled it by changing
> the yaml config `python-gateway.enabled : false` in api-server's configuration path `api-server/conf/application.yaml`
[jdk]: https://www.oracle.com/technetwork/java/javase/downloads/index.html
[zookeeper]: https://zookeeper.apache.org/releases.html

View File

@ -1,74 +0,0 @@
SkyWalking Agent Deployment
=============================
The `dolphinscheduler-skywalking` module provides [SkyWalking](https://skywalking.apache.org/) monitor agent for the DolphinScheduler project.
This document describes how to enable SkyWalking version 8.4+ support with this module (recommend using SkyWalking 8.5.0).
## Installation
The following configuration is used to enable the SkyWalking agent.
### Through Environment Variable Configuration (for Docker Compose)
Modify SkyWalking environment variables in `docker/docker-swarm/config.env.sh`:
```
SKYWALKING_ENABLE=true
SW_AGENT_COLLECTOR_BACKEND_SERVICES=127.0.0.1:11800
SW_GRPC_LOG_SERVER_HOST=127.0.0.1
SW_GRPC_LOG_SERVER_PORT=11800
```
And run:
```shell
$ docker-compose up -d
```
### Through Environment Variable Configuration (for Docker)
```shell
$ docker run -d --name dolphinscheduler \
-e DATABASE_HOST="192.168.x.x" -e DATABASE_PORT="5432" -e DATABASE_DATABASE="dolphinscheduler" \
-e DATABASE_USERNAME="test" -e DATABASE_PASSWORD="test" \
-e ZOOKEEPER_QUORUM="192.168.x.x:2181" \
-e SKYWALKING_ENABLE="true" \
-e SW_AGENT_COLLECTOR_BACKEND_SERVICES="your.skywalking-oap-server.com:11800" \
-e SW_GRPC_LOG_SERVER_HOST="your.skywalking-log-reporter.com" \
-e SW_GRPC_LOG_SERVER_PORT="11800" \
-p 12345:12345 \
apache/dolphinscheduler:1.3.8 all
```
### Through install_config.conf Configuration (for DolphinScheduler install.sh)
Add the following configurations to `${workDir}/conf/config/install_config.conf`.
```properties
# SkyWalking config
# note: enable SkyWalking tracking plugin
enableSkywalking="true"
# note: configure SkyWalking backend service address
skywalkingServers="your.skywalking-oap-server.com:11800"
# note: configure SkyWalking log reporter host
skywalkingLogReporterHost="your.skywalking-log-reporter.com"
# note: configure SkyWalking log reporter port
skywalkingLogReporterPort="11800"
```
## Usage
### Import Dashboard
#### Import DolphinScheduler Dashboard to SkyWalking Server
Copy the `${dolphinscheduler.home}/ext/skywalking-agent/dashboard/dolphinscheduler.yml` file into `${skywalking-oap-server.home}/config/ui-initialized-templates/` directory, and restart SkyWalking oap-server.
#### View DolphinScheduler Dashboard
If you have opened the SkyWalking dashboard with a browser before, you need to clear the browser cache.
![img1](/img/skywalking/import-dashboard-1.jpg)

View File

@ -39,4 +39,8 @@ sh ./bin/dolphinscheduler-daemon.sh start standalone-server
sh ./bin/dolphinscheduler-daemon.sh stop standalone-server
```
> Note: Python gateway service is started along with the api-server, and if you do not want to start Python gateway
> service please disabled it by changing the yaml config `python-gateway.enabled : false` in api-server's configuration
> path `api-server/conf/application.yaml`
[jdk]: https://www.oracle.com/technetwork/java/javase/downloads/index.html

View File

@ -65,4 +65,4 @@ Configure the required content according to the parameter descriptions above.
## Notice
JAVA and Scala only used for identification, there is no difference. If use Python to develop Flink, there is no class of the main function and the rest is the same.
JAVA and Scala only used for identification, there is no difference. If you use Python to develop Spark application, there is no class of the main function and the rest is the same.

View File

@ -6,7 +6,7 @@
`sh ./script/stop-all.sh`
## Download the Newest Version Installation Package
## Download the Latest Version Installation Package
- [download](/en-us/download/download.html) the latest version of the installation packages.
- The following upgrade operations need to be performed in the new version's directory.

View File

@ -523,18 +523,6 @@ A1edit /etc/nginx/conf.d/escheduler.conf
---
## Q欢迎订阅 DolphinScheduler 开发邮件列表
A在使用 DolphinScheduler 的过程中,如果您有任何问题或者想法、建议,都可以通过 Apache 邮件列表参与到 DolphinScheduler 的社区建设中来。
发送订阅邮件也非常简单,步骤如下:
1用自己的邮箱向 dev-subscribe@dolphinscheduler.apache.org 发送一封邮件,主题和内容任意。
2 接收确认邮件并回复。 完成步骤1后您将收到一封来自 dev-help@dolphinscheduler.apache.org 的确认邮件(如未收到,请确认邮件是否被自动归入垃圾邮件、推广邮件、订阅邮件等文件夹)。然后直接回复该邮件,或点击邮件里的链接快捷回复即可,主题和内容任意。
3 接收欢迎邮件。 完成以上步骤后,您会收到一封主题为 WELCOME to dev@dolphinscheduler.apache.org 的欢迎邮件,至此您已成功订阅 Apache DolphinScheduler的邮件列表。
---
## Q工作流依赖
A1目前是按照自然天来判断上月末判断时间是工作流 A start_time/scheduler_time between '2019-05-31 00:00:00' and '2019-05-31 23:59:59'。上月:是判断上个月从 1 号到月末每天都要有完成的A实例。上周 上周 7 天都要有完成的 A 实例。前两天: 判断昨天和前天,两天都要有完成的 A 实例。

View File

@ -1,74 +0,0 @@
SkyWalking Agent 部署
=============================
dolphinscheduler-skywalking 模块为 Dolphinscheduler 项目提供了 [Skywalking](https://skywalking.apache.org/) 监控代理。
本文档介绍了如何通过此模块接入 SkyWalking 8.4+ (推荐使用8.5.0)。
# 安装
以下配置用于启用 Skywalking agent。
### 通过配置环境变量 (使用 Docker Compose 部署时)
修改 `docker/docker-swarm/config.env.sh` 文件中的 SKYWALKING 环境变量:
```
SKYWALKING_ENABLE=true
SW_AGENT_COLLECTOR_BACKEND_SERVICES=127.0.0.1:11800
SW_GRPC_LOG_SERVER_HOST=127.0.0.1
SW_GRPC_LOG_SERVER_PORT=11800
```
并且运行
```shell
$ docker-compose up -d
```
### 通过配置环境变量 (使用 Docker 部署时)
```shell
$ docker run -d --name dolphinscheduler \
-e DATABASE_HOST="192.168.x.x" -e DATABASE_PORT="5432" -e DATABASE_DATABASE="dolphinscheduler" \
-e DATABASE_USERNAME="test" -e DATABASE_PASSWORD="test" \
-e ZOOKEEPER_QUORUM="192.168.x.x:2181" \
-e SKYWALKING_ENABLE="true" \
-e SW_AGENT_COLLECTOR_BACKEND_SERVICES="your.skywalking-oap-server.com:11800" \
-e SW_GRPC_LOG_SERVER_HOST="your.skywalking-log-reporter.com" \
-e SW_GRPC_LOG_SERVER_PORT="11800" \
-p 12345:12345 \
apache/dolphinscheduler:1.3.8 all
```
### 通过配置 install_config.conf (使用 DolphinScheduler install.sh 部署时)
添加以下配置到 `${workDir}/conf/config/install_config.conf`.
```properties
# skywalking config
# note: enable skywalking tracking plugin
enableSkywalking="true"
# note: configure skywalking backend service address
skywalkingServers="your.skywalking-oap-server.com:11800"
# note: configure skywalking log reporter host
skywalkingLogReporterHost="your.skywalking-log-reporter.com"
# note: configure skywalking log reporter port
skywalkingLogReporterPort="11800"
```
# 使用
### 导入图表
#### 导入图表到 Skywalking server
复制 `${dolphinscheduler.home}/ext/skywalking-agent/dashboard/dolphinscheduler.yml` 文件到 `${skywalking-oap-server.home}/config/ui-initialized-templates/` 目录下,并重启 Skywalking oap-server。
#### 查看 dolphinscheduler 图表
如果之前已经使用浏览器打开过 Skywalking则需要清空浏览器缓存。
![img1](/img/skywalking/import-dashboard-1.jpg)

Binary file not shown.

Before

Width:  |  Height:  |  Size: 148 KiB

View File

@ -32,16 +32,6 @@ root_dir: Path = Path(__file__).parent
img_dir: Path = root_dir.joinpath("img")
doc_dir: Path = root_dir.joinpath("docs")
expect_img_types: Set = {
"jpg",
"png",
}
def build_pattern() -> re.Pattern:
"""Build current document image regexp pattern."""
return re.compile(f"(/img.*?\\.({'|'.join(expect_img_types)}))")
def get_files_recurse(path: Path) -> Set:
"""Get all files recursively from given :param:`path`."""
@ -68,14 +58,15 @@ def get_paths_rel_path(paths: Set[Path], rel: Path) -> Set:
return {f"/{path.relative_to(rel)}" for path in paths}
def get_docs_img_path(paths: Set[Path], pattern: re.Pattern) -> Set:
def get_docs_img_path(paths: Set[Path]) -> Set:
"""Get all img syntax from given :param:`paths` using the regexp from :param:`pattern`."""
res = set()
pattern = re.compile(r"/img[\w./-]*")
for path in paths:
content = path.read_text()
find = pattern.findall(content)
if find:
res |= {item[0] for item in find}
res |= {item for item in find}
return res
@ -102,16 +93,6 @@ def diff_two_set(first: Set, second: Set) -> Tuple[set, set]:
return first.difference(second), second.difference(first)
def check_diff_img_type() -> Tuple[set, set]:
"""Check images difference type.
:return: Tuple[(actual - expect), (expect - actual)]
"""
img = get_files_recurse(img_dir)
img_suffix = get_paths_uniq_suffix(img)
return diff_two_set(img_suffix, expect_img_types)
def check_diff_img() -> Tuple[set, set]:
"""Check images difference files.
@ -120,20 +101,12 @@ def check_diff_img() -> Tuple[set, set]:
img = get_files_recurse(img_dir)
docs = get_files_recurse(doc_dir)
img_rel_path = get_paths_rel_path(img, root_dir)
pat = build_pattern()
docs_rel_path = get_docs_img_path(docs, pat)
docs_rel_path = get_docs_img_path(docs)
return diff_two_set(docs_rel_path, img_rel_path)
def check() -> None:
"""Runner for `check` sub command."""
img_type_act, img_type_exp = check_diff_img_type()
assert not img_type_act and not img_type_exp, (
f"Images type assert failed: \n"
f"* difference actual types to expect is: {img_type_act if img_type_act else 'None'}\n"
f"* difference expect types to actual is: {img_type_exp if img_type_exp else 'None'}\n"
)
img_docs, img_img = check_diff_img()
assert not img_docs and not img_img, (
f"Images assert failed: \n"

View File

@ -25,11 +25,13 @@ public enum ShowType {
* 1 TEXT;
* 2 attachment;
* 3 TABLE+attachment;
* 4 MARKDOWN;
*/
TABLE(0, "table"),
TEXT(1, "text"),
ATTACHMENT(2, "attachment"),
TABLE_ATTACHMENT(3, "table attachment");
TABLE_ATTACHMENT(3, "table attachment"),
MARKDOWN(4, "markdown"),;
private final int code;
private final String descp;

View File

@ -56,14 +56,14 @@ public final class WeChatAlertChannelFactory implements AlertChannelFactory {
.build();
InputParam usersParam = InputParam.newBuilder(WeChatAlertParamsConstants.NAME_ENTERPRISE_WE_CHAT_USERS, WeChatAlertParamsConstants.ENTERPRISE_WE_CHAT_USERS)
.setPlaceholder("please input users ")
.setPlaceholder("use `|` to separate userIds and `@all` to everyone ")
.addValidate(Validate.newBuilder()
.setRequired(true)
.setRequired(false)
.build())
.build();
InputParam agentIdParam = InputParam.newBuilder(WeChatAlertParamsConstants.NAME_ENTERPRISE_WE_CHAT_AGENT_ID, WeChatAlertParamsConstants.ENTERPRISE_WE_CHAT_AGENT_ID)
.setPlaceholder("please input agent id ")
.setPlaceholder("please input agent id or chat id ")
.addValidate(Validate.newBuilder()
.setRequired(true)
.build())
@ -77,9 +77,9 @@ public final class WeChatAlertChannelFactory implements AlertChannelFactory {
.build();
RadioParam showType = RadioParam.newBuilder(AlertConstants.NAME_SHOW_TYPE, AlertConstants.SHOW_TYPE)
.addParamsOptions(new ParamsOptions(ShowType.TABLE.getDescp(), ShowType.TABLE.getDescp(), false))
.addParamsOptions(new ParamsOptions(ShowType.MARKDOWN.getDescp(), ShowType.MARKDOWN.getDescp(), false))
.addParamsOptions(new ParamsOptions(ShowType.TEXT.getDescp(), ShowType.TEXT.getDescp(), false))
.setValue(ShowType.TABLE.getDescp())
.setValue(ShowType.MARKDOWN.getDescp())
.addValidate(Validate.newBuilder().setRequired(true).build())
.build();

View File

@ -24,9 +24,8 @@ public final class WeChatAlertParamsConstants {
static final String NAME_ENTERPRISE_WE_CHAT_SECRET = "secret";
static final String ENTERPRISE_WE_CHAT_TEAM_SEND_MSG = "$t('teamSendMsg')";
static final String NAME_ENTERPRISE_WE_CHAT_TEAM_SEND_MSG = "teamSendMsg";
static final String ENTERPRISE_WE_CHAT_AGENT_ID = "$t('agentId')";
static final String NAME_ENTERPRISE_WE_CHAT_AGENT_ID = "agentId";
static final String NAME_ENTERPRISE_WE_CHAT_CHAT_ID = "chatId";
static final String ENTERPRISE_WE_CHAT_AGENT_ID = "$t('agentId/chatId')";
static final String NAME_ENTERPRISE_WE_CHAT_AGENT_ID = "agentId/chatId";
static final String ENTERPRISE_WE_CHAT_USERS = "$t('users')";
static final String NAME_ENTERPRISE_WE_CHAT_USERS = "users";

View File

@ -17,15 +17,10 @@
package org.apache.dolphinscheduler.plugin.alert.wechat;
import static java.util.Objects.requireNonNull;
import static org.apache.dolphinscheduler.plugin.alert.wechat.WeChatAlertConstants.*;
import org.apache.dolphinscheduler.alert.api.AlertConstants;
import org.apache.dolphinscheduler.alert.api.AlertResult;
import org.apache.dolphinscheduler.alert.api.ShowType;
import org.apache.dolphinscheduler.spi.utils.JSONUtils;
import org.apache.dolphinscheduler.spi.utils.StringUtils;
import org.apache.http.HttpEntity;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpGet;
@ -34,18 +29,20 @@ import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import java.io.IOException;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import static java.util.Objects.requireNonNull;
import static org.apache.dolphinscheduler.plugin.alert.wechat.WeChatAlertConstants.*;
public final class WeChatSender {
private static final Logger logger = LoggerFactory.getLogger(WeChatSender.class);
@ -57,8 +54,7 @@ public final class WeChatSender {
private static final String CORP_ID_REGEX = "{corpId}";
private static final String SECRET_REGEX = "{secret}";
private static final String TOKEN_REGEX = "{token}";
private final String weChatAgentId;
private final String weChatChatId;
private final String weChatAgentIdChatId;
private final String weChatUsers;
private final String weChatTokenUrlReplace;
private final String weChatToken;
@ -66,8 +62,7 @@ public final class WeChatSender {
private final String showType;
WeChatSender(Map<String, String> config) {
weChatAgentId = config.get(WeChatAlertParamsConstants.NAME_ENTERPRISE_WE_CHAT_AGENT_ID);
weChatChatId = config.get(WeChatAlertParamsConstants.NAME_ENTERPRISE_WE_CHAT_CHAT_ID);
weChatAgentIdChatId = config.get(WeChatAlertParamsConstants.NAME_ENTERPRISE_WE_CHAT_AGENT_ID);
weChatUsers = config.get(WeChatAlertParamsConstants.NAME_ENTERPRISE_WE_CHAT_USERS);
String weChatCorpId = config.get(WeChatAlertParamsConstants.NAME_ENTERPRISE_WE_CHAT_CORP_ID);
String weChatSecret = config.get(WeChatAlertParamsConstants.NAME_ENTERPRISE_WE_CHAT_SECRET);
@ -76,8 +71,8 @@ public final class WeChatSender {
showType = config.get(AlertConstants.NAME_SHOW_TYPE);
requireNonNull(showType, AlertConstants.NAME_SHOW_TYPE + MUST_NOT_NULL);
weChatTokenUrlReplace = weChatTokenUrl
.replace(CORP_ID_REGEX, weChatCorpId)
.replace(SECRET_REGEX, weChatSecret);
.replace(CORP_ID_REGEX, weChatCorpId)
.replace(SECRET_REGEX, weChatSecret);
weChatToken = getToken();
}
@ -100,42 +95,10 @@ public final class WeChatSender {
}
}
/**
* convert table to markdown style
*
* @param title the title
* @param content the content
* @return markdown table content
*/
private static String markdownTable(String title, String content) {
List<LinkedHashMap> mapItemsList = JSONUtils.toList(content, LinkedHashMap.class);
if (null == mapItemsList || mapItemsList.isEmpty()) {
logger.error("itemsList is null");
throw new RuntimeException("itemsList is null");
}
StringBuilder contents = new StringBuilder(200);
for (LinkedHashMap mapItems : mapItemsList) {
Set<Entry<String, Object>> entries = mapItems.entrySet();
Iterator<Entry<String, Object>> iterator = entries.iterator();
StringBuilder t = new StringBuilder(String.format("`%s`%s", title, WeChatAlertConstants.MARKDOWN_ENTER));
while (iterator.hasNext()) {
Map.Entry<String, Object> entry = iterator.next();
t.append(WeChatAlertConstants.MARKDOWN_QUOTE);
t.append(entry.getKey()).append(":").append(entry.getValue());
t.append(WeChatAlertConstants.MARKDOWN_ENTER);
}
contents.append(t);
}
return contents.toString();
}
/**
* convert text to markdown style
*
* @param title the title
* @param title the title
* @param content the content
* @return markdown text
*/
@ -242,17 +205,17 @@ public final class WeChatSender {
return alertResult;
}
String enterpriseWeChatPushUrlReplace = "";
Map<String,String> contentMap=new HashMap<>();
contentMap.put(WeChatAlertConstants.WE_CHAT_CONTENT_KEY,data);
String msgJson="";
Map<String, String> contentMap = new HashMap<>();
contentMap.put(WeChatAlertConstants.WE_CHAT_CONTENT_KEY, data);
String msgJson = "";
if (sendType.equals(WeChatType.APP.getDescp())) {
enterpriseWeChatPushUrlReplace = WeChatAlertConstants.WE_CHAT_PUSH_URL.replace(TOKEN_REGEX, weChatToken);
WechatAppMessage wechatAppMessage=new WechatAppMessage(weChatUsers, WE_CHAT_MESSAGE_TYPE_TEXT, Integer.valueOf(weChatAgentId),contentMap, WE_CHAT_MESSAGE_SAFE_PUBLICITY, WE_CHAT_ENABLE_ID_TRANS, WE_CHAT_DUPLICATE_CHECK_INTERVAL_ZERO);
msgJson=JSONUtils.toJsonString(wechatAppMessage);
WechatAppMessage wechatAppMessage = new WechatAppMessage(weChatUsers, showType, Integer.valueOf(weChatAgentIdChatId), contentMap, WE_CHAT_MESSAGE_SAFE_PUBLICITY, WE_CHAT_ENABLE_ID_TRANS, WE_CHAT_DUPLICATE_CHECK_INTERVAL_ZERO);
msgJson = JSONUtils.toJsonString(wechatAppMessage);
} else if (sendType.equals(WeChatType.APPCHAT.getDescp())) {
enterpriseWeChatPushUrlReplace = WeChatAlertConstants.WE_CHAT_APP_CHAT_PUSH_URL.replace(TOKEN_REGEX, weChatToken);
WechatAppChatMessage wechatAppChatMessage=new WechatAppChatMessage(weChatChatId, WE_CHAT_MESSAGE_TYPE_TEXT, contentMap, WE_CHAT_MESSAGE_SAFE_PUBLICITY);
msgJson=JSONUtils.toJsonString(wechatAppChatMessage);
WechatAppChatMessage wechatAppChatMessage = new WechatAppChatMessage(weChatAgentIdChatId, showType, contentMap, WE_CHAT_MESSAGE_SAFE_PUBLICITY);
msgJson = JSONUtils.toJsonString(wechatAppChatMessage);
}
try {
@ -272,14 +235,7 @@ public final class WeChatSender {
* @return the markdown alert table/text
*/
private String markdownByAlert(String title, String content) {
String result = "";
if (showType.equals(ShowType.TABLE.getDescp())) {
result = markdownTable(title, content);
} else if (showType.equals(ShowType.TEXT.getDescp())) {
result = markdownText(title, content);
}
return result;
return markdownText(title, content);
}
private String getToken() {

View File

@ -17,6 +17,8 @@
package org.apache.dolphinscheduler.plugin.alert.wechat;
import org.apache.dolphinscheduler.alert.api.ShowType;
import java.util.Map;
public class WechatAppChatMessage {
@ -24,6 +26,7 @@ public class WechatAppChatMessage {
private String chatid;
private String msgtype;
private Map<String,String> text;
private Map<String,String> markdown;
private Integer safe;
public String getChatid() {
@ -58,13 +61,25 @@ public class WechatAppChatMessage {
this.safe = safe;
}
public Map<String, String> getMarkdown() {
return markdown;
}
public void setMarkdown(Map<String, String> markdown) {
this.markdown = markdown;
}
public WechatAppChatMessage() {
}
public WechatAppChatMessage(String chatid, String msgtype, Map<String, String> text, Integer safe) {
public WechatAppChatMessage(String chatid, String msgtype, Map<String, String> contentMap, Integer safe) {
this.chatid = chatid;
this.msgtype = msgtype;
this.text = text;
if (msgtype.equals(ShowType.MARKDOWN.getDescp())) {
this.markdown = contentMap;
} else {
this.text = contentMap;
}
this.safe = safe;
}
}

View File

@ -17,6 +17,8 @@
package org.apache.dolphinscheduler.plugin.alert.wechat;
import org.apache.dolphinscheduler.alert.api.ShowType;
import java.util.Map;
public class WechatAppMessage {
@ -24,7 +26,8 @@ public class WechatAppMessage {
private String touser;
private String msgtype;
private Integer agentid;
private Map<String,String> text;
private Map<String, String> text;
private Map<String, String> markdown;
private Integer safe;
private Integer enable_id_trans;
private Integer enable_duplicate_check;
@ -85,16 +88,28 @@ public class WechatAppMessage {
this.enable_duplicate_check = enable_duplicate_check;
}
public Map<String, String> getMarkdown() {
return markdown;
}
public void setMarkdown(Map<String, String> markdown) {
this.markdown = markdown;
}
public WechatAppMessage() {
}
public WechatAppMessage(String touser, String msgtype, Integer agentid, Map<String, String> text, Integer safe, Integer enable_id_trans, Integer enable_duplicate_check) {
public WechatAppMessage(String touser, String msgtype, Integer agentid, Map<String, String> contentMap, Integer safe, Integer enableIdTrans, Integer enableDuplicateCheck) {
this.touser = touser;
this.msgtype = msgtype;
this.agentid = agentid;
this.text = text;
if (msgtype.equals(ShowType.MARKDOWN.getDescp())) {
this.markdown = contentMap;
} else {
this.text = contentMap;
}
this.safe = safe;
this.enable_id_trans = enable_id_trans;
this.enable_duplicate_check = enable_duplicate_check;
this.enable_id_trans = enableIdTrans;
this.enable_duplicate_check = enableDuplicateCheck;
}
}

View File

@ -41,8 +41,6 @@ import java.util.Optional;
import java.util.ServiceLoader;
import java.util.Set;
import javax.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.context.event.ApplicationReadyEvent;
@ -55,23 +53,23 @@ public final class AlertPluginManager {
private final PluginDao pluginDao;
private final Map<Integer, AlertChannel> channelKeyedById = new HashMap<>();
private final PluginParams warningTypeParams = getWarningTypeParams();
public AlertPluginManager(PluginDao pluginDao) {
this.pluginDao = pluginDao;
}
private final Map<Integer, AlertChannel> channelKeyedById = new HashMap<>();
private final PluginParams warningTypeParams = getWarningTypeParams();
public PluginParams getWarningTypeParams() {
return
RadioParam.newBuilder(AlertConstants.NAME_WARNING_TYPE, AlertConstants.WARNING_TYPE)
.addParamsOptions(new ParamsOptions(WarningType.SUCCESS.getDescp(), WarningType.SUCCESS.getDescp(), false))
.addParamsOptions(new ParamsOptions(WarningType.FAILURE.getDescp(), WarningType.FAILURE.getDescp(), false))
.addParamsOptions(new ParamsOptions(WarningType.ALL.getDescp(), WarningType.ALL.getDescp(), false))
.setValue(WarningType.ALL.getDescp())
.addValidate(Validate.newBuilder().setRequired(true).build())
.build();
RadioParam.newBuilder(AlertConstants.NAME_WARNING_TYPE, AlertConstants.WARNING_TYPE)
.addParamsOptions(new ParamsOptions(WarningType.SUCCESS.getDescp(), WarningType.SUCCESS.getDescp(), false))
.addParamsOptions(new ParamsOptions(WarningType.FAILURE.getDescp(), WarningType.FAILURE.getDescp(), false))
.addParamsOptions(new ParamsOptions(WarningType.ALL.getDescp(), WarningType.ALL.getDescp(), false))
.setValue(WarningType.ALL.getDescp())
.addValidate(Validate.newBuilder().setRequired(true).build())
.build();
}
@EventListener

View File

@ -36,10 +36,10 @@ import io.netty.channel.Channel;
public final class AlertRequestProcessor implements NettyRequestProcessor {
private static final Logger logger = LoggerFactory.getLogger(AlertRequestProcessor.class);
private final AlertSender alertSender;
private final AlertSenderService alertSenderService;
public AlertRequestProcessor(AlertSender alertSender) {
this.alertSender = alertSender;
public AlertRequestProcessor(AlertSenderService alertSenderService) {
this.alertSenderService = alertSenderService;
}
@Override
@ -51,7 +51,7 @@ public final class AlertRequestProcessor implements NettyRequestProcessor {
logger.info("Received command : {}", alertSendRequestCommand);
AlertSendResponseCommand alertSendResponseCommand = alertSender.syncHandler(
AlertSendResponseCommand alertSendResponseCommand = alertSenderService.syncHandler(
alertSendRequestCommand.getGroupId(),
alertSendRequestCommand.getTitle(),
alertSendRequestCommand.getContent(),

View File

@ -17,43 +17,65 @@
package org.apache.dolphinscheduler.alert;
import org.apache.commons.collections.CollectionUtils;
import org.apache.dolphinscheduler.alert.api.AlertChannel;
import org.apache.dolphinscheduler.alert.api.AlertConstants;
import org.apache.dolphinscheduler.alert.api.AlertData;
import org.apache.dolphinscheduler.alert.api.AlertInfo;
import org.apache.dolphinscheduler.alert.api.AlertResult;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.AlertStatus;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.dao.entity.Alert;
import org.apache.dolphinscheduler.dao.entity.AlertPluginInstance;
import org.apache.dolphinscheduler.remote.command.alert.AlertSendResponseCommand;
import org.apache.dolphinscheduler.remote.command.alert.AlertSendResponseResult;
import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;
@Component
public final class AlertSender {
private static final Logger logger = LoggerFactory.getLogger(AlertSender.class);
@Service
public final class AlertSenderService extends Thread {
private static final Logger logger = LoggerFactory.getLogger(AlertSenderService.class);
private final AlertDao alertDao;
private final AlertPluginManager alertPluginManager;
public AlertSender(AlertDao alertDao, AlertPluginManager alertPluginManager) {
public AlertSenderService(AlertDao alertDao, AlertPluginManager alertPluginManager) {
this.alertDao = alertDao;
this.alertPluginManager = alertPluginManager;
}
@Override
public synchronized void start() {
super.setName("AlertSenderService");
super.start();
}
@Override
public void run() {
logger.info("alert sender started");
while (Stopper.isRunning()) {
try {
List<Alert> alerts = alertDao.listPendingAlerts();
this.send(alerts);
ThreadUtils.sleep(Constants.SLEEP_TIME_MILLIS * 5L);
} catch (Exception e) {
logger.error("alert sender thread error", e);
}
}
}
public void send(List<Alert> alerts) {
for (Alert alert : alerts) {
//get alert group from alert
@ -66,11 +88,11 @@ public final class AlertSender {
}
AlertData alertData = new AlertData();
alertData.setId(alert.getId())
.setContent(alert.getContent())
.setLog(alert.getLog())
.setTitle(alert.getTitle())
.setTitle(alert.getTitle())
.setWarnType(alert.getWarningType().getCode());
.setContent(alert.getContent())
.setLog(alert.getLog())
.setTitle(alert.getTitle())
.setTitle(alert.getTitle())
.setWarnType(alert.getWarningType().getCode());
for (AlertPluginInstance instance : alertInstanceList) {
@ -81,23 +103,22 @@ public final class AlertSender {
}
}
}
}
/**
* sync send alert handler
*
* @param alertGroupId alertGroupId
* @param title title
* @param content content
* @param title title
* @param content content
* @return AlertSendResponseCommand
*/
public AlertSendResponseCommand syncHandler(int alertGroupId, String title, String content , int warnType) {
public AlertSendResponseCommand syncHandler(int alertGroupId, String title, String content, int warnType) {
List<AlertPluginInstance> alertInstanceList = alertDao.listInstanceByAlertGroupId(alertGroupId);
AlertData alertData = new AlertData();
alertData.setContent(content)
.setTitle(title)
.setWarnType(warnType);
.setTitle(title)
.setWarnType(warnType);
boolean sendResponseStatus = true;
List<AlertSendResponseResult> sendResponseResults = new ArrayList<>();
@ -116,7 +137,7 @@ public final class AlertSender {
AlertResult alertResult = this.alertResultHandler(instance, alertData);
if (alertResult != null) {
AlertSendResponseResult alertSendResponseResult = new AlertSendResponseResult(
Boolean.parseBoolean(String.valueOf(alertResult.getStatus())), alertResult.getMessage());
Boolean.parseBoolean(String.valueOf(alertResult.getStatus())), alertResult.getMessage());
sendResponseStatus = sendResponseStatus && alertSendResponseResult.getStatus();
sendResponseResults.add(alertSendResponseResult);
}
@ -128,7 +149,7 @@ public final class AlertSender {
/**
* alert result handler
*
* @param instance instance
* @param instance instance
* @param alertData alertData
* @return AlertResult
*/
@ -147,7 +168,7 @@ public final class AlertSender {
Map<String, String> paramsMap = JSONUtils.toMap(instance.getPluginInstanceParams());
String instanceWarnType = WarningType.ALL.getDescp();
if(paramsMap != null){
if (paramsMap != null) {
instanceWarnType = paramsMap.getOrDefault(AlertConstants.NAME_WARNING_TYPE, WarningType.ALL.getDescp());
}

View File

@ -17,73 +17,95 @@
package org.apache.dolphinscheduler.alert;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.thread.Stopper;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.common.thread.ThreadUtils;
import org.apache.dolphinscheduler.dao.PluginDao;
import org.apache.dolphinscheduler.dao.entity.Alert;
import org.apache.dolphinscheduler.remote.NettyRemotingServer;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
import java.io.Closeable;
import java.util.List;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
import javax.annotation.PreDestroy;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.WebApplicationType;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.builder.SpringApplicationBuilder;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.context.event.EventListener;
import javax.annotation.PreDestroy;
import java.io.Closeable;
@SpringBootApplication
@ComponentScan("org.apache.dolphinscheduler")
public class AlertServer implements Closeable {
private static final Logger logger = LoggerFactory.getLogger(AlertServer.class);
private final PluginDao pluginDao;
private final AlertDao alertDao;
private final AlertPluginManager alertPluginManager;
private final AlertSender alertSender;
private final AlertSenderService alertSenderService;
private final AlertRequestProcessor alertRequestProcessor;
private final AlertConfig alertConfig;
private NettyRemotingServer nettyRemotingServer;
private NettyRemotingServer server;
@Autowired
private AlertConfig config;
public AlertServer(PluginDao pluginDao, AlertDao alertDao, AlertPluginManager alertPluginManager, AlertSender alertSender, AlertRequestProcessor alertRequestProcessor) {
public AlertServer(PluginDao pluginDao, AlertSenderService alertSenderService, AlertRequestProcessor alertRequestProcessor, AlertConfig alertConfig) {
this.pluginDao = pluginDao;
this.alertDao = alertDao;
this.alertPluginManager = alertPluginManager;
this.alertSender = alertSender;
this.alertSenderService = alertSenderService;
this.alertRequestProcessor = alertRequestProcessor;
this.alertConfig = alertConfig;
}
/**
* alert server startup, not use web service
*
* @param args arguments
*/
public static void main(String[] args) {
SpringApplication.run(AlertServer.class, args);
Thread.currentThread().setName(Constants.THREAD_NAME_ALERT_SERVER);
new SpringApplicationBuilder(AlertServer.class).web(WebApplicationType.NONE).run(args);
}
@EventListener
public void start(ApplicationReadyEvent readyEvent) {
logger.info("Starting Alert server");
public void run(ApplicationReadyEvent readyEvent) {
logger.info("alert server starting...");
checkTable();
startServer();
Executors.newScheduledThreadPool(1)
.scheduleAtFixedRate(new Sender(), 5, 5, TimeUnit.SECONDS);
alertSenderService.start();
}
@Override
@PreDestroy
public void close() {
server.close();
destroy("alert server destroy");
}
/**
* gracefully stop
*
* @param cause stop cause
*/
public void destroy(String cause) {
try {
// execute only once
if (Stopper.isStopped()) {
return;
}
logger.info("alert server is stopping ..., cause : {}", cause);
// set stop signal is true
Stopper.stop();
// thread sleep 3 seconds for thread quietly stop
ThreadUtils.sleep(3000L);
// close
this.nettyRemotingServer.close();
} catch (Exception e) {
logger.error("alert server stop exception ", e);
}
}
private void checkTable() {
@ -95,26 +117,11 @@ public class AlertServer implements Closeable {
private void startServer() {
NettyServerConfig serverConfig = new NettyServerConfig();
serverConfig.setListenPort(config.getPort());
serverConfig.setListenPort(alertConfig.getPort());
server = new NettyRemotingServer(serverConfig);
server.registerProcessor(CommandType.ALERT_SEND_REQUEST, alertRequestProcessor);
server.start();
nettyRemotingServer = new NettyRemotingServer(serverConfig);
nettyRemotingServer.registerProcessor(CommandType.ALERT_SEND_REQUEST, alertRequestProcessor);
nettyRemotingServer.start();
}
final class Sender implements Runnable {
@Override
public void run() {
if (!Stopper.isRunning()) {
return;
}
try {
final List<Alert> alerts = alertDao.listPendingAlerts();
alertSender.send(alerts);
} catch (Exception e) {
logger.error("Failed to send alert", e);
}
}
}
}

View File

@ -18,7 +18,9 @@
package org.apache.dolphinscheduler.alert;
import junit.framework.TestCase;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.dao.PluginDao;
import org.apache.dolphinscheduler.dao.entity.Alert;
import org.apache.dolphinscheduler.remote.NettyRemotingServer;
import org.apache.dolphinscheduler.remote.config.NettyServerConfig;
import org.junit.Assert;
@ -27,9 +29,13 @@ import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
import org.mockito.junit.MockitoJUnitRunner;
import org.powermock.reflect.Whitebox;
import java.util.ArrayList;
import java.util.List;
@RunWith(MockitoJUnitRunner.class)
public class AlertServerTest extends TestCase {
@ -42,19 +48,26 @@ public class AlertServerTest extends TestCase {
@Mock
private AlertConfig alertConfig;
@Mock
private AlertSenderService alertSenderService;
@Test
public void testStart() {
Mockito.when(pluginDao.checkPluginDefineTableExist()).thenReturn(true);
Mockito.when(alertConfig.getPort()).thenReturn(50053);
alertServer.start(null);
Mockito.doNothing().when(alertSenderService).start();
alertServer.run(null);
NettyRemotingServer nettyRemotingServer = Whitebox.getInternalState(alertServer, "server");
NettyRemotingServer nettyRemotingServer = Whitebox.getInternalState(alertServer, "nettyRemotingServer");
NettyServerConfig nettyServerConfig = Whitebox.getInternalState(nettyRemotingServer, "serverConfig");
Assert.assertEquals(50053, nettyServerConfig.getListenPort());
}
}

View File

@ -20,30 +20,35 @@ package org.apache.dolphinscheduler.alert.processor;
import static org.mockito.Mockito.mock;
import org.apache.dolphinscheduler.alert.AlertRequestProcessor;
import org.apache.dolphinscheduler.alert.AlertSender;
import org.apache.dolphinscheduler.alert.AlertSenderService;
import org.apache.dolphinscheduler.common.enums.WarningType;
import org.apache.dolphinscheduler.dao.AlertDao;
import org.apache.dolphinscheduler.remote.command.Command;
import org.apache.dolphinscheduler.remote.command.CommandType;
import org.apache.dolphinscheduler.remote.command.alert.AlertSendRequestCommand;
import org.apache.dolphinscheduler.remote.command.alert.AlertSendResponseCommand;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import io.netty.channel.Channel;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.junit.MockitoJUnitRunner;
@RunWith(MockitoJUnitRunner.class)
public class AlertRequestProcessorTest {
@InjectMocks
private AlertRequestProcessor alertRequestProcessor;
@Before
public void before() {
final AlertDao alertDao = mock(AlertDao.class);
alertRequestProcessor = new AlertRequestProcessor(new AlertSender(alertDao, null));
}
@Mock
private AlertSenderService alertSenderService;
@Test
public void testProcess() {
Mockito.when(alertSenderService.syncHandler(1, "title", "content", WarningType.FAILURE.getCode())).thenReturn(new AlertSendResponseCommand());
Channel channel = mock(Channel.class);
AlertSendRequestCommand alertSendRequestCommand = new AlertSendRequestCommand(1, "title", "content", WarningType.FAILURE.getCode());
Command reqCommand = alertSendRequestCommand.convert2Command();

View File

@ -21,7 +21,7 @@ import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
import org.apache.dolphinscheduler.alert.AlertPluginManager;
import org.apache.dolphinscheduler.alert.AlertSender;
import org.apache.dolphinscheduler.alert.AlertSenderService;
import org.apache.dolphinscheduler.alert.api.AlertChannel;
import org.apache.dolphinscheduler.alert.api.AlertResult;
import org.apache.dolphinscheduler.common.enums.WarningType;
@ -39,24 +39,29 @@ import java.util.Optional;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
import org.mockito.InjectMocks;
import org.mockito.Mock;
import org.mockito.Mockito;
import org.mockito.MockitoAnnotations;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
public class AlertSenderTest {
private static final Logger logger = LoggerFactory.getLogger(AlertSenderTest.class);
public class AlertSenderServiceTest {
private static final Logger logger = LoggerFactory.getLogger(AlertSenderServiceTest.class);
@Mock
private AlertDao alertDao;
@Mock
private PluginDao pluginDao;
@Mock
private AlertPluginManager alertPluginManager;
private AlertSender alertSender;
@InjectMocks
private AlertSenderService alertSenderService;
@Before
public void before() {
alertDao = mock(AlertDao.class);
pluginDao = mock(PluginDao.class);
alertPluginManager = mock(AlertPluginManager.class);
MockitoAnnotations.initMocks(this);
}
@Test
@ -65,12 +70,11 @@ public class AlertSenderTest {
int alertGroupId = 1;
String title = "alert mail test title";
String content = "alert mail test content";
alertSender = new AlertSender(alertDao, alertPluginManager);
//1.alert instance does not exist
when(alertDao.listInstanceByAlertGroupId(alertGroupId)).thenReturn(null);
AlertSendResponseCommand alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
AlertSendResponseCommand alertSendResponseCommand = alertSenderService.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
@ -89,7 +93,7 @@ public class AlertSenderTest {
PluginDefine pluginDefine = new PluginDefine(pluginName, "1", null);
when(pluginDao.getPluginDefineById(pluginDefineId)).thenReturn(pluginDefine);
alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
alertSendResponseCommand = alertSenderService.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
@ -99,7 +103,7 @@ public class AlertSenderTest {
when(alertChannelMock.process(Mockito.any())).thenReturn(null);
when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock));
alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
alertSendResponseCommand = alertSenderService.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
@ -111,7 +115,7 @@ public class AlertSenderTest {
when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult);
when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock));
alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
alertSendResponseCommand = alertSenderService.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
Assert.assertFalse(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
@ -123,7 +127,7 @@ public class AlertSenderTest {
when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult);
when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock));
alertSendResponseCommand = alertSender.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
alertSendResponseCommand = alertSenderService.syncHandler(alertGroupId, title, content, WarningType.ALL.getCode());
Assert.assertTrue(alertSendResponseCommand.getResStatus());
alertSendResponseCommand.getResResults().forEach(result ->
logger.info("alert send response result, status:{}, message:{}", result.getStatus(), result.getMessage()));
@ -143,7 +147,7 @@ public class AlertSenderTest {
alert.setWarningType(WarningType.FAILURE);
alertList.add(alert);
alertSender = new AlertSender(alertDao, alertPluginManager);
// alertSenderService = new AlertSenderService();
int pluginDefineId = 1;
String pluginInstanceParams = "alert-instance-mail-params";
@ -165,6 +169,7 @@ public class AlertSenderTest {
when(alertChannelMock.process(Mockito.any())).thenReturn(alertResult);
when(alertPluginManager.getAlertChannel(1)).thenReturn(Optional.of(alertChannelMock));
Assert.assertTrue(Boolean.parseBoolean(alertResult.getStatus()));
alertSender.send(alertList);
when(alertDao.listInstanceByAlertGroupId(1)).thenReturn(new ArrayList<>());
alertSenderService.send(alertList);
}
}

View File

@ -275,6 +275,12 @@
</exclusions>
<scope>test</scope>
</dependency>
<!-- Python -->
<dependency>
<groupId>net.sf.py4j</groupId>
<artifactId>py4j</artifactId>
</dependency>
</dependencies>
<build>

View File

@ -29,6 +29,6 @@ WORKDIR $DOLPHINSCHEDULER_HOME
ADD ./target/api-server $DOLPHINSCHEDULER_HOME
EXPOSE 12345
EXPOSE 12345 25333
CMD [ "/bin/bash", "./bin/start.sh" ]

View File

@ -15,17 +15,17 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.server.config;
package org.apache.dolphinscheduler.api.configuration;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Configuration;
import org.springframework.stereotype.Component;
@Component
@EnableConfigurationProperties
@ConfigurationProperties("python-gateway")
public class PythonGatewayConfig {
@ConfigurationProperties(value = "python-gateway", ignoreUnknownFields = false)
public class PythonGatewayConfiguration {
private boolean enabled;
private String gatewayServerAddress;
private int gatewayServerPort;
private String pythonAddress;
@ -33,6 +33,14 @@ public class PythonGatewayConfig {
private int connectTimeout;
private int readTimeout;
public boolean getEnabled() {
return enabled;
}
public void setEnabled(boolean enabled) {
this.enabled = enabled;
}
public String getGatewayServerAddress() {
return gatewayServerAddress;
}

View File

@ -95,22 +95,22 @@ public class ExecutorController extends BaseController {
*/
@ApiOperation(value = "startProcessInstance", notes = "RUN_PROCESS_INSTANCE_NOTES")
@ApiImplicitParams({
@ApiImplicitParam(name = "processDefinitionCode", value = "PROCESS_DEFINITION_CODE", required = true, dataType = "Long", example = "100"),
@ApiImplicitParam(name = "scheduleTime", value = "SCHEDULE_TIME", dataType = "String"),
@ApiImplicitParam(name = "failureStrategy", value = "FAILURE_STRATEGY", required = true, dataType = "FailureStrategy"),
@ApiImplicitParam(name = "startNodeList", value = "START_NODE_LIST", dataType = "String"),
@ApiImplicitParam(name = "taskDependType", value = "TASK_DEPEND_TYPE", dataType = "TaskDependType"),
@ApiImplicitParam(name = "execType", value = "COMMAND_TYPE", dataType = "CommandType"),
@ApiImplicitParam(name = "warningType", value = "WARNING_TYPE", required = true, dataType = "WarningType"),
@ApiImplicitParam(name = "warningGroupId", value = "WARNING_GROUP_ID", dataType = "Int", example = "100"),
@ApiImplicitParam(name = "runMode", value = "RUN_MODE", dataType = "RunMode"),
@ApiImplicitParam(name = "processInstancePriority", value = "PROCESS_INSTANCE_PRIORITY", required = true, dataType = "Priority"),
@ApiImplicitParam(name = "workerGroup", value = "WORKER_GROUP", dataType = "String", example = "default"),
@ApiImplicitParam(name = "environmentCode", value = "ENVIRONMENT_CODE", dataType = "Long", example = "-1"),
@ApiImplicitParam(name = "timeout", value = "TIMEOUT", dataType = "Int", example = "100"),
@ApiImplicitParam(name = "expectedParallelismNumber", value = "EXPECTED_PARALLELISM_NUMBER", dataType = "Int" , example = "8"),
@ApiImplicitParam(name = "dryRun", value = "DRY_RUN", dataType = "Int", example = "0"),
@ApiImplicitParam(name = "complementDependentMode", value = "COMPLEMENT_DEPENDENT_MODE", dataType = "complementDependentMode")
@ApiImplicitParam(name = "processDefinitionCode", value = "PROCESS_DEFINITION_CODE", required = true, dataType = "Long", example = "100"),
@ApiImplicitParam(name = "scheduleTime", value = "SCHEDULE_TIME", required = true, dataType = "String", example = "2022-04-06 00:00:00,2022-04-06 00:00:00"),
@ApiImplicitParam(name = "failureStrategy", value = "FAILURE_STRATEGY", required = true, dataType = "FailureStrategy"),
@ApiImplicitParam(name = "startNodeList", value = "START_NODE_LIST", dataType = "String"),
@ApiImplicitParam(name = "taskDependType", value = "TASK_DEPEND_TYPE", dataType = "TaskDependType"),
@ApiImplicitParam(name = "execType", value = "COMMAND_TYPE", dataType = "CommandType"),
@ApiImplicitParam(name = "warningType", value = "WARNING_TYPE", required = true, dataType = "WarningType"),
@ApiImplicitParam(name = "warningGroupId", value = "WARNING_GROUP_ID", dataType = "Int", example = "100"),
@ApiImplicitParam(name = "runMode", value = "RUN_MODE", dataType = "RunMode"),
@ApiImplicitParam(name = "processInstancePriority", value = "PROCESS_INSTANCE_PRIORITY", required = true, dataType = "Priority"),
@ApiImplicitParam(name = "workerGroup", value = "WORKER_GROUP", dataType = "String", example = "default"),
@ApiImplicitParam(name = "environmentCode", value = "ENVIRONMENT_CODE", dataType = "Long", example = "-1"),
@ApiImplicitParam(name = "timeout", value = "TIMEOUT", dataType = "Int", example = "100"),
@ApiImplicitParam(name = "expectedParallelismNumber", value = "EXPECTED_PARALLELISM_NUMBER", dataType = "Int" , example = "8"),
@ApiImplicitParam(name = "dryRun", value = "DRY_RUN", dataType = "Int", example = "0"),
@ApiImplicitParam(name = "complementDependentMode", value = "COMPLEMENT_DEPENDENT_MODE", dataType = "complementDependentMode")
})
@PostMapping(value = "start-process-instance")
@ResponseStatus(HttpStatus.OK)
@ -119,7 +119,7 @@ public class ExecutorController extends BaseController {
public Result startProcessInstance(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@ApiParam(name = "projectCode", value = "PROJECT_CODE", required = true) @PathVariable long projectCode,
@RequestParam(value = "processDefinitionCode") long processDefinitionCode,
@RequestParam(value = "scheduleTime", required = false) String scheduleTime,
@RequestParam(value = "scheduleTime") String scheduleTime,
@RequestParam(value = "failureStrategy") FailureStrategy failureStrategy,
@RequestParam(value = "startNodeList", required = false) String startNodeList,
@RequestParam(value = "taskDependType", required = false) TaskDependType taskDependType,
@ -159,7 +159,7 @@ public class ExecutorController extends BaseController {
* batch execute process instance
* If any processDefinitionCode cannot be found, the failure information is returned and the status is set to
* failed. The successful task will run normally and will not stop
*
*
* @param loginUser login user
* @param projectCode project code
* @param processDefinitionCodes process definition codes
@ -180,7 +180,7 @@ public class ExecutorController extends BaseController {
@ApiOperation(value = "batchStartProcessInstance", notes = "BATCH_RUN_PROCESS_INSTANCE_NOTES")
@ApiImplicitParams({
@ApiImplicitParam(name = "processDefinitionCodes", value = "PROCESS_DEFINITION_CODES", required = true, dataType = "String", example = "1,2,3"),
@ApiImplicitParam(name = "scheduleTime", value = "SCHEDULE_TIME", required = true, dataType = "String"),
@ApiImplicitParam(name = "scheduleTime", value = "SCHEDULE_TIME", required = true, dataType = "String", example = "2022-04-06 00:00:00,2022-04-06 00:00:00"),
@ApiImplicitParam(name = "failureStrategy", value = "FAILURE_STRATEGY", required = true, dataType = "FailureStrategy"),
@ApiImplicitParam(name = "startNodeList", value = "START_NODE_LIST", dataType = "String"),
@ApiImplicitParam(name = "taskDependType", value = "TASK_DEPEND_TYPE", dataType = "TaskDependType"),
@ -201,24 +201,24 @@ public class ExecutorController extends BaseController {
@ApiException(START_PROCESS_INSTANCE_ERROR)
@AccessLogAnnotation(ignoreRequestArgs = "loginUser")
public Result batchStartProcessInstance(@ApiIgnore @RequestAttribute(value = Constants.SESSION_USER) User loginUser,
@ApiParam(name = "projectCode", value = "PROJECT_CODE", required = true) @PathVariable long projectCode,
@RequestParam(value = "processDefinitionCodes") String processDefinitionCodes,
@RequestParam(value = "scheduleTime", required = false) String scheduleTime,
@RequestParam(value = "failureStrategy") FailureStrategy failureStrategy,
@RequestParam(value = "startNodeList", required = false) String startNodeList,
@RequestParam(value = "taskDependType", required = false) TaskDependType taskDependType,
@RequestParam(value = "execType", required = false) CommandType execType,
@RequestParam(value = "warningType") WarningType warningType,
@RequestParam(value = "warningGroupId", required = false) int warningGroupId,
@RequestParam(value = "runMode", required = false) RunMode runMode,
@RequestParam(value = "processInstancePriority", required = false) Priority processInstancePriority,
@RequestParam(value = "workerGroup", required = false, defaultValue = "default") String workerGroup,
@RequestParam(value = "environmentCode", required = false, defaultValue = "-1") Long environmentCode,
@RequestParam(value = "timeout", required = false) Integer timeout,
@RequestParam(value = "startParams", required = false) String startParams,
@RequestParam(value = "expectedParallelismNumber", required = false) Integer expectedParallelismNumber,
@RequestParam(value = "dryRun", defaultValue = "0", required = false) int dryRun,
@RequestParam(value = "complementDependentMode", required = false) ComplementDependentMode complementDependentMode) {
@ApiParam(name = "projectCode", value = "PROJECT_CODE", required = true) @PathVariable long projectCode,
@RequestParam(value = "processDefinitionCodes") String processDefinitionCodes,
@RequestParam(value = "scheduleTime") String scheduleTime,
@RequestParam(value = "failureStrategy") FailureStrategy failureStrategy,
@RequestParam(value = "startNodeList", required = false) String startNodeList,
@RequestParam(value = "taskDependType", required = false) TaskDependType taskDependType,
@RequestParam(value = "execType", required = false) CommandType execType,
@RequestParam(value = "warningType") WarningType warningType,
@RequestParam(value = "warningGroupId", required = false) int warningGroupId,
@RequestParam(value = "runMode", required = false) RunMode runMode,
@RequestParam(value = "processInstancePriority", required = false) Priority processInstancePriority,
@RequestParam(value = "workerGroup", required = false, defaultValue = "default") String workerGroup,
@RequestParam(value = "environmentCode", required = false, defaultValue = "-1") Long environmentCode,
@RequestParam(value = "timeout", required = false) Integer timeout,
@RequestParam(value = "startParams", required = false) String startParams,
@RequestParam(value = "expectedParallelismNumber", required = false) Integer expectedParallelismNumber,
@RequestParam(value = "dryRun", defaultValue = "0", required = false) int dryRun,
@RequestParam(value = "complementDependentMode", required = false) ComplementDependentMode complementDependentMode) {
if (timeout == null) {
timeout = Constants.MAX_TASK_TIMEOUT;
@ -269,8 +269,8 @@ public class ExecutorController extends BaseController {
*/
@ApiOperation(value = "execute", notes = "EXECUTE_ACTION_TO_PROCESS_INSTANCE_NOTES")
@ApiImplicitParams({
@ApiImplicitParam(name = "processInstanceId", value = "PROCESS_INSTANCE_ID", required = true, dataType = "Int", example = "100"),
@ApiImplicitParam(name = "executeType", value = "EXECUTE_TYPE", required = true, dataType = "ExecuteType")
@ApiImplicitParam(name = "processInstanceId", value = "PROCESS_INSTANCE_ID", required = true, dataType = "Int", example = "100"),
@ApiImplicitParam(name = "executeType", value = "EXECUTE_TYPE", required = true, dataType = "ExecuteType")
})
@PostMapping(value = "/execute")
@ResponseStatus(HttpStatus.OK)
@ -293,7 +293,7 @@ public class ExecutorController extends BaseController {
*/
@ApiOperation(value = "startCheckProcessDefinition", notes = "START_CHECK_PROCESS_DEFINITION_NOTES")
@ApiImplicitParams({
@ApiImplicitParam(name = "processDefinitionCode", value = "PROCESS_DEFINITION_CODE", required = true, dataType = "Long", example = "100")
@ApiImplicitParam(name = "processDefinitionCode", value = "PROCESS_DEFINITION_CODE", required = true, dataType = "Long", example = "100")
})
@PostMapping(value = "/start-check")
@ResponseStatus(HttpStatus.OK)

View File

@ -15,7 +15,7 @@
* limitations under the License.
*/
package org.apache.dolphinscheduler.server;
package org.apache.dolphinscheduler.api.python;
import org.apache.dolphinscheduler.api.dto.resources.ResourceComponent;
import org.apache.dolphinscheduler.api.enums.Status;
@ -56,7 +56,7 @@ import org.apache.dolphinscheduler.dao.mapper.ProjectMapper;
import org.apache.dolphinscheduler.dao.mapper.ProjectUserMapper;
import org.apache.dolphinscheduler.dao.mapper.ScheduleMapper;
import org.apache.dolphinscheduler.dao.mapper.TaskDefinitionMapper;
import org.apache.dolphinscheduler.server.config.PythonGatewayConfig;
import org.apache.dolphinscheduler.api.configuration.PythonGatewayConfiguration;
import org.apache.dolphinscheduler.spi.enums.ResourceType;
import org.apache.commons.collections.CollectionUtils;
@ -75,17 +75,13 @@ import javax.annotation.PostConstruct;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;
import org.springframework.boot.web.servlet.support.SpringBootServletInitializer;
import org.springframework.context.annotation.ComponentScan;
import org.springframework.stereotype.Component;
import py4j.GatewayServer;
@SpringBootApplication
@ComponentScan(value = "org.apache.dolphinscheduler")
public class PythonGatewayServer extends SpringBootServletInitializer {
private static final Logger logger = LoggerFactory.getLogger(PythonGatewayServer.class);
@Component
public class PythonGateway {
private static final Logger logger = LoggerFactory.getLogger(PythonGateway.class);
private static final WarningType DEFAULT_WARNING_TYPE = WarningType.NONE;
private static final int DEFAULT_WARNING_GROUP_ID = 0;
@ -141,7 +137,7 @@ public class PythonGatewayServer extends SpringBootServletInitializer {
private DataSourceMapper dataSourceMapper;
@Autowired
private PythonGatewayConfig pythonGatewayConfig;
private PythonGatewayConfiguration pythonGatewayConfiguration;
@Autowired
private ProjectUserMapper projectUserMapper;
@ -546,30 +542,32 @@ public class PythonGatewayServer extends SpringBootServletInitializer {
}
@PostConstruct
public void run() {
GatewayServer server;
try {
InetAddress gatewayHost = InetAddress.getByName(pythonGatewayConfig.getGatewayServerAddress());
InetAddress pythonHost = InetAddress.getByName(pythonGatewayConfig.getPythonAddress());
server = new GatewayServer(
this,
pythonGatewayConfig.getGatewayServerPort(),
pythonGatewayConfig.getPythonPort(),
gatewayHost,
pythonHost,
pythonGatewayConfig.getConnectTimeout(),
pythonGatewayConfig.getReadTimeout(),
null
);
GatewayServer.turnLoggingOn();
logger.info("PythonGatewayServer started on: " + gatewayHost.toString());
server.start();
} catch (UnknownHostException e) {
logger.error("exception occurred while constructing PythonGatewayServer().", e);
public void init() {
if (pythonGatewayConfiguration.getEnabled()) {
this.start();
}
}
public static void main(String[] args) {
SpringApplication.run(PythonGatewayServer.class, args);
private void start() {
GatewayServer server;
try {
InetAddress gatewayHost = InetAddress.getByName(pythonGatewayConfiguration.getGatewayServerAddress());
InetAddress pythonHost = InetAddress.getByName(pythonGatewayConfiguration.getPythonAddress());
server = new GatewayServer(
this,
pythonGatewayConfiguration.getGatewayServerPort(),
pythonGatewayConfiguration.getPythonPort(),
gatewayHost,
pythonHost,
pythonGatewayConfiguration.getConnectTimeout(),
pythonGatewayConfiguration.getReadTimeout(),
null
);
GatewayServer.turnLoggingOn();
logger.info("PythonGatewayService started on: " + gatewayHost.toString());
server.start();
} catch (UnknownHostException e) {
logger.error("exception occurred while constructing PythonGatewayService().", e);
}
}
}

View File

@ -619,7 +619,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
* @param runMode
* @return
*/
private int createComplementCommandList(Date start, Date end, RunMode runMode, Command command,
protected int createComplementCommandList(Date start, Date end, RunMode runMode, Command command,
Integer expectedParallelismNumber, ComplementDependentMode complementDependentMode) {
int createCount = 0;
int dependentProcessDefinitionCreateCount = 0;
@ -713,7 +713,7 @@ public class ExecutorServiceImpl extends BaseServiceImpl implements ExecutorServ
/**
* create complement dependent command
*/
private int createComplementDependentCommand(List<Schedule> schedules, Command command) {
protected int createComplementDependentCommand(List<Schedule> schedules, Command command) {
int dependentProcessDefinitionCreateCount = 0;
Command dependentCommand;

View File

@ -214,7 +214,6 @@ public class ProcessInstanceServiceImpl extends BaseServiceImpl implements Proce
if (processDefinition == null || projectCode != processDefinition.getProjectCode()) {
putMsg(result, Status.PROCESS_DEFINE_NOT_EXIST, processId);
} else {
processInstance.setWarningGroupId(processDefinition.getWarningGroupId());
processInstance.setLocations(processDefinition.getLocations());
processInstance.setDagData(processService.genDagData(processDefinition));
result.put(DATA_LIST, processInstance);

View File

@ -108,6 +108,26 @@ audit:
metrics:
enabled: true
python-gateway:
# Weather enable python gateway server or not. The default value is true.
enabled: true
# The address of Python gateway server start. Set its value to `0.0.0.0` if your Python API run in different
# between Python gateway server. It could be be specific to other address like `127.0.0.1` or `localhost`
gateway-server-address: 0.0.0.0
# The port of Python gateway server start. Define which port you could connect to Python gateway server from
# Python API side.
gateway-server-port: 25333
# The address of Python callback client.
python-address: 127.0.0.1
# The port of Python callback client.
python-port: 25334
# Close connection of socket server if no other request accept after x milliseconds. Define value is (0 = infinite),
# and socket server would never close even though no requests accept
connect-timeout: 0
# Close each active connection of socket server if python program not active after x milliseconds. Define value is
# (0 = infinite), and socket server would never close even though no requests accept
read-timeout: 0
# Override by profile
---

View File

@ -219,7 +219,7 @@ QUERY_ALL_DEFINITION_LIST_NOTES=query all definition list
PAGE_NO=page no
PROCESS_INSTANCE_ID=process instance id
PROCESS_INSTANCE_JSON=process instance info(json format)
SCHEDULE_TIME=schedule time
SCHEDULE_TIME=schedule time,empty string indicates the current day
SYNC_DEFINE=update the information of the process instance to the process definition
RECOVERY_PROCESS_INSTANCE_FLAG=whether to recovery process instance
PREVIEW_SCHEDULE_NOTES=preview schedule

View File

@ -206,7 +206,7 @@ PROCESS_INSTANCE_ID=流程实例ID
PROCESS_INSTANCE_IDS=流程实例ID集合
PROCESS_INSTANCE_JSON=流程实例信息(json格式)
PREVIEW_SCHEDULE_NOTES=定时调度预览
SCHEDULE_TIME=定时时间
SCHEDULE_TIME=定时时间,空字符串表示当前天
SYNC_DEFINE=更新流程实例的信息是否同步到流程定义
RECOVERY_PROCESS_INSTANCE_FLAG=是否恢复流程实例
SEARCH_VAL=搜索值

View File

@ -202,9 +202,10 @@ public class ExecutorControllerTest extends AbstractControllerTest {
paramsMap.add("processDefinitionCode", String.valueOf(processDefinitionCode));
paramsMap.add("failureStrategy", String.valueOf(failureStrategy));
paramsMap.add("warningType", String.valueOf(warningType));
paramsMap.add("scheduleTime", scheduleTime);
when(executorService.execProcessInstance(any(User.class), eq(projectCode), eq(processDefinitionCode),
eq(null), eq(null), eq(failureStrategy), eq(null), eq(null), eq(warningType),
eq(scheduleTime), eq(null), eq(failureStrategy), eq(null), eq(null), eq(warningType),
eq(0), eq(null), eq(null), eq("default"), eq(-1L),
eq(Constants.MAX_TASK_TIMEOUT), eq(null), eq(null), eq(0),
eq(complementDependentMode))).thenReturn(executeServiceResult);

View File

@ -327,6 +327,7 @@ public final class Constants {
public static final String NULL = "NULL";
public static final String THREAD_NAME_MASTER_SERVER = "Master-Server";
public static final String THREAD_NAME_WORKER_SERVER = "Worker-Server";
public static final String THREAD_NAME_ALERT_SERVER = "Alert-Server";
/**
* command parameter keys

View File

@ -17,19 +17,18 @@
package org.apache.dolphinscheduler.dao.entity;
import org.apache.dolphinscheduler.common.enums.UdfType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import org.apache.commons.lang.StringUtils;
import java.io.IOException;
import java.util.Date;
import com.baomidou.mybatisplus.annotation.IdType;
import com.baomidou.mybatisplus.annotation.TableField;
import com.baomidou.mybatisplus.annotation.TableId;
import com.baomidou.mybatisplus.annotation.TableName;
import com.fasterxml.jackson.databind.DeserializationContext;
import com.fasterxml.jackson.databind.KeyDeserializer;
import org.apache.commons.lang.StringUtils;
import org.apache.dolphinscheduler.common.enums.UdfType;
import org.apache.dolphinscheduler.common.utils.JSONUtils;
import java.io.IOException;
import java.util.Date;
/**
* udf function
@ -39,13 +38,23 @@ public class UdfFunc {
/**
* id
*/
@TableId(value="id", type=IdType.AUTO)
@TableId(value = "id", type = IdType.AUTO)
private int id;
/**
* user id
*/
private int userId;
public String getResourceType() {
return resourceType;
}
public void setResourceType(String resourceType) {
this.resourceType = "UDF";
}
@TableField(exist = false)
private String resourceType = "UDF";
/**
* udf function name
*/

View File

@ -18,10 +18,11 @@
package org.apache.dolphinscheduler.dao.entity;
import org.apache.dolphinscheduler.dao.entity.UdfFunc.UdfFuncDeserializer;
import java.io.IOException;
import org.junit.Assert;
import org.junit.Test;
import java.io.IOException;
public class UdfFuncTest {
/**
@ -35,9 +36,9 @@ public class UdfFuncTest {
udfFunc.setResourceId(2);
udfFunc.setClassName("org.apache.dolphinscheduler.test.mrUpdate");
Assert.assertEquals("{\"id\":0,\"userId\":0,\"funcName\":null,\"className\":\"org.apache.dolphinscheduler.test.mrUpdate\",\"argTypes\":null,\"database\":null,"
+ "\"description\":null,\"resourceId\":2,\"resourceName\":\"dolphin_resource_update\",\"type\":null,\"createTime\":null,\"updateTime\":null}"
, udfFunc.toString());
Assert.assertEquals("{\"id\":0,\"userId\":0,\"resourceType\":\"UDF\",\"funcName\":null,\"className\":\"org.apache.dolphinscheduler.test.mrUpdate\",\"argTypes\":null,\"database\":null,"
+ "\"description\":null,\"resourceId\":2,\"resourceName\":\"dolphin_resource_update\",\"type\":null,\"createTime\":null,\"updateTime\":null}"
, udfFunc.toString());
}
/**

View File

@ -50,11 +50,6 @@
<outputDirectory>logger-server</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-python/target/python-gateway-server</directory>
<outputDirectory>python-gateway-server</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-standalone-server/target/standalone-server</directory>
<outputDirectory>standalone-server</outputDirectory>

View File

@ -112,10 +112,7 @@ public class MasterRegistryClient {
}
public void start() {
String nodeLock = Constants.REGISTRY_DOLPHINSCHEDULER_LOCK_FAILOVER_STARTUP_MASTERS;
try {
// create distributed lock with the root node path of the lock space as /dolphinscheduler/lock/failover/startup-masters
registryClient.getLock(nodeLock);
// master registry
registry();
@ -123,8 +120,6 @@ public class MasterRegistryClient {
} catch (Exception e) {
logger.error("master start up exception", e);
throw new RuntimeException("master start up error", e);
} finally {
registryClient.releaseLock(nodeLock);
}
}

View File

@ -28,82 +28,7 @@
<name>${project.artifactId}</name>
<packaging>jar</packaging>
<dependencies>
<!-- dolphinscheduler -->
<dependency>
<groupId>org.apache.dolphinscheduler</groupId>
<artifactId>dolphinscheduler-api</artifactId>
</dependency>
<!--springboot-->
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-web</artifactId>
<exclusions>
<exclusion>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-tomcat</artifactId>
</exclusion>
<exclusion>
<artifactId>log4j-to-slf4j</artifactId>
<groupId>org.apache.logging.log4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>net.sf.py4j</groupId>
<artifactId>py4j</artifactId>
</dependency>
</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-jar-plugin</artifactId>
<configuration>
<excludes>
<exclude>*.yaml</exclude>
<exclude>*.xml</exclude>
</excludes>
</configuration>
</plugin>
<plugin>
<artifactId>maven-assembly-plugin</artifactId>
<executions>
<execution>
<id>dolphinscheduler-python-gateway-server</id>
<phase>package</phase>
<goals>
<goal>single</goal>
</goals>
<configuration>
<finalName>python-gateway-server</finalName>
<descriptors>
<descriptor>src/main/assembly/dolphinscheduler-python-gateway-server.xml</descriptor>
</descriptors>
<appendAssemblyId>false</appendAssemblyId>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
<profiles>
<profile>
<id>docker</id>
<build>
<plugins>
<plugin>
<groupId>org.codehaus.mojo</groupId>
<artifactId>exec-maven-plugin</artifactId>
</plugin>
</plugins>
</build>
</profile>
<profile>
<id>python</id>
<build>

View File

@ -24,6 +24,9 @@ It started after version 2.0.5 released
## dev
* Integrate Python gateway server into Dolphinscheduler API server, and you could start Python gateway service by command
`./bin/dolphinscheduler-daemon.sh start api-server` instead of independent command
`./bin/dolphinscheduler-daemon.sh start python-gateway-server`.
* Remove parameter `queue` from class `ProcessDefinition` to avoid confuse user when it change but not work
* Change `yaml_parser.py` method `to_string` to magic method `__str__` make it more pythonic.
* Use package ``ruamel.yaml`` replace ``pyyaml`` for write yaml file with comment.

View File

@ -55,7 +55,7 @@ After Python is already installed on your machine following section
$ pip install apache-dolphinscheduler
The latest version of *PyDolphinScheduler* would be installed after you run above
command in your terminal. You could go and `start Python Gateway Server`_ to finish
command in your terminal. You could go and `start Python Gateway Service`_ to finish
the prepare, and then go to :doc:`tutorial` to make your hand dirty. But if you
want to install the unreleased version of *PyDolphinScheduler*, you could go and see
section `installing PyDolphinScheduler in dev`_ for more detail.
@ -74,33 +74,39 @@ which we hold in GitHub
# Install PyDolphinScheduler in develop mode
$ cd dolphinscheduler-python/pydolphinscheduler && pip install -e .
After you installed *PyDolphinScheduler*, please remember `start Python Gateway Server`_
After you installed *PyDolphinScheduler*, please remember `start Python Gateway Service`_
which waiting for *PyDolphinScheduler*'s workflow definition require.
Start Python Gateway Server
---------------------------
Start Python Gateway Service
----------------------------
Since **PyDolphinScheduler** is Python API for `Apache DolphinScheduler`_, it
could define workflow and tasks structure, but could not run it unless you
`install Apache DolphinScheduler`_ and start Python gateway server. We only
and some key steps here and you could go `install Apache DolphinScheduler`_
for more detail
`install Apache DolphinScheduler`_ and start its API server which including
Python gateway service in it. We only and some key steps here and you could
go `install Apache DolphinScheduler`_ for more detail
.. code-block:: bash
# Start pythonGatewayServer
$ ./bin/dolphinscheduler-daemon.sh start pythonGatewayServer
# Start DolphinScheduler api-server which including python gateway service
$ ./bin/dolphinscheduler-daemon.sh start api-server
To check whether the server is alive or not, you could run :code:`jps`. And
the server is health if keyword `PythonGatewayServer` in the console.
the server is health if keyword `ApiApplicationServer` in the console.
.. code-block:: bash
$ jps
....
201472 PythonGatewayServer
201472 ApiApplicationServer
....
.. note::
Please make sure you already enabled started Python gateway service along with `api-server`. The configuration is in
yaml config path `python-gateway.enabled : true` in api-server's configuration path in `api-server/conf/application.yaml`.
The default value is true and Python gateway service start when api server is been started.
What's More
-----------

View File

@ -130,7 +130,7 @@ Now, we could run the Python code like other Python script, for the basic usage
:end-before: [end submit_or_run]
If you not start your Apache DolphinScheduler server, you could find the way in
:ref:`start:start Python gateway server` and it would have more detail about related server
:ref:`start:start Python gateway service` and it would have more detail about related server
start. Beside attribute `run`, we have attribute `submit` for object `ProcessDefinition`
and it just submit workflow to the daemon but not setting the schedule information. For
more detail you could see :ref:`concept:process definition`.

View File

@ -15,4 +15,4 @@
# specific language governing permissions and limitations
# under the License.
"""Test integration between Python API and PythonGatewayServer."""
"""Test integration between Python API and PythonGatewayService."""

View File

@ -15,7 +15,7 @@
# specific language governing permissions and limitations
# under the License.
"""Test whether success submit examples DAG to PythonGatewayServer."""
"""Test whether success submit examples DAG to PythonGatewayService."""
from pathlib import Path
@ -34,7 +34,7 @@ from tests.testing.path import path_example
],
)
def test_exec_white_list_example(example_path: Path):
"""Test execute examples and submit DAG to PythonGatewayServer."""
"""Test execute examples and submit DAG to PythonGatewayService."""
try:
exec(example_path.read_text())
except Exception:

View File

@ -1,64 +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.
-->
<assembly xmlns="http://maven.apache.org/ASSEMBLY/2.1.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/ASSEMBLY/2.1.0 http://maven.apache.org/xsd/assembly-2.1.0.xsd">
<id>dolphinscheduler-python-gateway-server</id>
<formats>
<format>dir</format>
</formats>
<includeBaseDirectory>false</includeBaseDirectory>
<baseDirectory>python-gateway-server</baseDirectory>
<fileSets>
<fileSet>
<directory>${basedir}/src/main/resources</directory>
<includes>
<include>*.yaml</include>
<include>*.xml</include>
</includes>
<outputDirectory>conf</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/src/main/bin</directory>
<outputDirectory>bin</outputDirectory>
<fileMode>0755</fileMode>
<directoryMode>0755</directoryMode>
</fileSet>
<fileSet>
<directory>${basedir}/../script/env</directory>
<outputDirectory>bin</outputDirectory>
<includes>
<include>dolphinscheduler_env.sh</include>
</includes>
<fileMode>0755</fileMode>
<directoryMode>0755</directoryMode>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-common/src/main/resources</directory>
<includes>
<include>**/*.properties</include>
</includes>
<outputDirectory>conf</outputDirectory>
</fileSet>
</fileSets>
<dependencySets>
<dependencySet>
<outputDirectory>libs</outputDirectory>
</dependencySet>
</dependencySets>
</assembly>

View File

@ -1,32 +0,0 @@
#!/bin/bash
#
# 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.
#
BIN_DIR=$(dirname $0)
DOLPHINSCHEDULER_HOME=${DOLPHINSCHEDULER_HOME:-$(cd $BIN_DIR/..; pwd)}
source "$BIN_DIR/dolphinscheduler_env.sh"
JAVA_OPTS=${JAVA_OPTS:-"-server -Duser.timezone=${SPRING_JACKSON_TIME_ZONE} -Xms1g -Xmx1g -Xmn512m -XX:+PrintGCDetails -Xloggc:gc.log -XX:+HeapDumpOnOutOfMemoryError -XX:HeapDumpPath=dump.hprof"}
if [[ "$DOCKER" == "true" ]]; then
JAVA_OPTS="${JAVA_OPTS} -XX:-UseContainerSupport"
fi
java $JAVA_OPTS \
-cp "$DOLPHINSCHEDULER_HOME/conf":"$DOLPHINSCHEDULER_HOME/libs/*" \
org.apache.dolphinscheduler.server.PythonGatewayServer

View File

@ -1,34 +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.
#
FROM openjdk:8-jre-slim-buster
ENV DOCKER true
ENV TZ Asia/Shanghai
ENV DOLPHINSCHEDULER_HOME /opt/dolphinscheduler
RUN apt update ; \
apt install -y curl sudo ; \
rm -rf /var/lib/apt/lists/*
WORKDIR $DOLPHINSCHEDULER_HOME
ADD ./target/python-gateway-server $DOLPHINSCHEDULER_HOME
EXPOSE 25333 54321
CMD [ "/bin/bash", "./bin/start.sh" ]

View File

@ -1,83 +0,0 @@
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License. You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
spring:
application:
name: python-gateway-server
main:
banner-mode: off
datasource:
driver-class-name: org.h2.Driver
url: jdbc:h2:mem:dolphinscheduler;MODE=MySQL;DB_CLOSE_DELAY=-1;DATABASE_TO_LOWER=true
username: sa
password: ""
jackson:
time-zone: UTC
date-format: "yyyy-MM-dd HH:mm:ss"
servlet:
multipart:
max-file-size: 1024MB
max-request-size: 1024MB
messages:
basename: i18n/messages
jpa:
hibernate:
ddl-auto: none
python-gateway:
# The address of Python gateway server start. Set its value to `0.0.0.0` if your Python API run in different
# between Python gateway server. It could be be specific to other address like `127.0.0.1` or `localhost`
gateway-server-address: 0.0.0.0
# The port of Python gateway server start. Define which port you could connect to Python gateway server from
# Python API side.
gateway-server-port: 25333
# The address of Python callback client.
python-address: 127.0.0.1
# The port of Python callback client.
python-port: 25334
# Close connection of socket server if no other request accept after x milliseconds. Define value is (0 = infinite),
# and socket server would never close even though no requests accept
connect-timeout: 0
# Close each active connection of socket server if python program not active after x milliseconds. Define value is
# (0 = infinite), and socket server would never close even though no requests accept
read-timeout: 0
server:
port: 54321
management:
endpoints:
web:
exposure:
include: '*'
metrics:
tags:
application: ${spring.application.name}
metrics:
enabled: true
# Override by profile
---
spring:
config:
activate:
on-profile: postgresql
quartz:
properties:
org.quartz.jobStore.driverDelegateClass: org.quartz.impl.jdbcjobstore.PostgreSQLDelegate

View File

@ -1,57 +0,0 @@
<?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.
-->
<configuration scan="true" scanPeriod="120 seconds">
<property name="log.base" value="logs"/>
<appender name="STDOUT" class="ch.qos.logback.core.ConsoleAppender">
<encoder>
<pattern>
[%level] %date{yyyy-MM-dd HH:mm:ss.SSS} %logger{96}:[%line] - %msg%n
</pattern>
<charset>UTF-8</charset>
</encoder>
</appender>
<appender name="PYTHONGATEWAYLOGFILE" class="ch.qos.logback.core.rolling.RollingFileAppender">
<file>${log.base}/dolphinscheduler-python-gateway.log</file>
<filter class="ch.qos.logback.classic.filter.ThresholdFilter">
<level>INFO</level>
</filter>
<rollingPolicy class="ch.qos.logback.core.rolling.SizeAndTimeBasedRollingPolicy">
<fileNamePattern>${log.base}/dolphinscheduler-python-gateway.%d{yyyy-MM-dd_HH}.%i.log</fileNamePattern>
<maxHistory>168</maxHistory>
<maxFileSize>64MB</maxFileSize>
</rollingPolicy>
<encoder>
<pattern>
[%level] %date{yyyy-MM-dd HH:mm:ss.SSS} %logger{96}:[%line] - %msg%n
</pattern>
<charset>UTF-8</charset>
</encoder>
</appender>
<root level="INFO">
<if condition="${DOCKER:-false}">
<then>
<appender-ref ref="STDOUT"/>
</then>
</if>
<appender-ref ref="PYTHONGATEWAYLOGFILE"/>
</root>
</configuration>

View File

@ -88,4 +88,14 @@ public class AlertSendRequestCommand implements Serializable {
command.setBody(body);
return command;
}
@Override
public String toString() {
return "AlertSendRequestCommand{" +
"groupId=" + groupId +
", title='" + title + '\'' +
", content='" + content + '\'' +
", warnType=" + warnType +
'}';
}
}

View File

@ -17,12 +17,7 @@
package org.apache.dolphinscheduler.service.process;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_PARAMS;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_CODE;
import static org.mockito.ArgumentMatchers.any;
import com.fasterxml.jackson.databind.JsonNode;
import org.apache.dolphinscheduler.common.Constants;
import org.apache.dolphinscheduler.common.enums.CommandType;
import org.apache.dolphinscheduler.common.enums.Flag;
@ -78,14 +73,6 @@ import org.apache.dolphinscheduler.plugin.task.api.model.ResourceInfo;
import org.apache.dolphinscheduler.service.exceptions.ServiceException;
import org.apache.dolphinscheduler.service.quartz.cron.CronUtilsTest;
import org.apache.dolphinscheduler.spi.params.base.FormType;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.junit.Assert;
import org.junit.Rule;
import org.junit.Test;
@ -99,7 +86,17 @@ import org.powermock.reflect.Whitebox;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import com.fasterxml.jackson.databind.JsonNode;
import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_RECOVER_PROCESS_ID_STRING;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_START_PARAMS;
import static org.apache.dolphinscheduler.common.Constants.CMD_PARAM_SUB_PROCESS_DEFINE_CODE;
import static org.mockito.ArgumentMatchers.any;
/**
* process service test
@ -113,7 +110,7 @@ public class ProcessServiceTest {
public final ExpectedException exception = ExpectedException.none();
@InjectMocks
private ProcessService processService;
private ProcessServiceImpl processService;
@Mock
private CommandMapper commandMapper;
@Mock
@ -850,8 +847,8 @@ public class ProcessServiceTest {
int pageNumber = 0;
int masterCount = 0;
int thisMasterSlot = 2;
List<Command> commandList = processService.findCommandPageBySlot(pageSize,pageNumber,masterCount,thisMasterSlot);
Assert.assertEquals(0,commandList.size());
List<Command> commandList = processService.findCommandPageBySlot(pageSize, pageNumber, masterCount, thisMasterSlot);
Assert.assertEquals(0, commandList.size());
}
private TaskGroupQueue getTaskGroupQueue() {

View File

@ -45,10 +45,6 @@
<directory>${basedir}/../dolphinscheduler-log-server/target/logger-server/libs</directory>
<outputDirectory>libs/logger-server</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/../dolphinscheduler-python/target/python-gateway-server/libs</directory>
<outputDirectory>libs/python-gateway</outputDirectory>
</fileSet>
<fileSet>
<directory>${basedir}/src/main/resources</directory>

View File

@ -28,7 +28,7 @@ if [[ "$DOCKER" == "true" ]]; then
fi
CP=$DOLPHINSCHEDULER_HOME/libs/standalone-server/*
for d in alert-server api-server master-server python-gateway-server worker-server; do
for d in alert-server api-server master-server worker-server; do
for f in $DOLPHINSCHEDULER_HOME/../$d/libs/*.jar; do
CP=$CP:$f
done

View File

@ -140,6 +140,8 @@ alert:
port: 50052
python-gateway:
# Weather enable python gateway server or not. The default value is true.
enabled: true
# The address of Python gateway server start. Set its value to `0.0.0.0` if your Python API run in different
# between Python gateway server. It could be be specific to other address like `127.0.0.1` or `localhost`
gateway-server-address: 0.0.0.0

View File

@ -75,15 +75,14 @@ public abstract class AbstractParameters implements IParameters {
* @return parameters map
*/
public Map<String, Property> getLocalParametersMap() {
Map<String, Property> localParametersMaps = new LinkedHashMap<>();
if (localParams != null) {
Map<String, Property> localParametersMaps = new LinkedHashMap<>();
for (Property property : localParams) {
localParametersMaps.put(property.getProp(),property);
}
return localParametersMaps;
}
return null;
return localParametersMaps;
}
/**
@ -92,14 +91,13 @@ public abstract class AbstractParameters implements IParameters {
* @return parameters map
*/
public Map<String, Property> getVarPoolMap() {
Map<String, Property> varPoolMap = new LinkedHashMap<>();
if (varPool != null) {
Map<String, Property> varPoolMap = new LinkedHashMap<>();
for (Property property : varPool) {
varPoolMap.put(property.getProp(), property);
}
return varPoolMap;
}
return null;
return varPoolMap;
}
public List<Property> getVarPool() {

View File

@ -21,11 +21,12 @@ import com.fasterxml.jackson.annotation.JsonSubTypes;
import com.fasterxml.jackson.annotation.JsonSubTypes.Type;
import com.fasterxml.jackson.annotation.JsonTypeInfo;
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, property = "type")
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME, visible = true, property = "resourceType")
@JsonSubTypes({
@Type(value = DataSourceParameters.class, name = "DATASOURCE"),
@Type(value = UdfFuncParameters.class, name = "UDF")
})
public abstract class AbstractResourceParameters {
}

View File

@ -17,12 +17,24 @@
package org.apache.dolphinscheduler.plugin.task.api.parameters.resource;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.dolphinscheduler.spi.enums.DbType;
public class DataSourceParameters extends AbstractResourceParameters {
private DbType type;
public String getResourceType() {
return resourceType;
}
public void setResourceType(String resourceType) {
this.resourceType = resourceType;
}
@JsonProperty(value = "DATASOURCE")
private String resourceType;
private String connectionParams;
public DbType getType() {

View File

@ -17,6 +17,7 @@
package org.apache.dolphinscheduler.plugin.task.api.parameters.resource;
import com.fasterxml.jackson.annotation.JsonProperty;
import org.apache.dolphinscheduler.plugin.task.api.enums.UdfType;
import org.apache.dolphinscheduler.spi.utils.JSONUtils;
@ -30,6 +31,18 @@ public class UdfFuncParameters extends AbstractResourceParameters {
* id
*/
private int id;
public String getResourceType() {
return resourceType;
}
public void setResourceType(String resourceType) {
this.resourceType = resourceType;
}
@JsonProperty(value = "UDF")
private String resourceType;
/**
* user id
*/

View File

@ -63,9 +63,10 @@ public class ParamUtils {
// combining local and global parameters
Map<String, Property> localParams = parameters.getLocalParametersMap();
//stream pass params
Map<String, Property> varParams = parameters.getVarPoolMap();
if (globalParams == null && localParams == null) {
if (globalParams.size() == 0 && localParams.size() == 0 && varParams.size() == 0) {
return null;
}
// if it is a complement,
@ -85,15 +86,13 @@ public class ParamUtils {
}
params.put(PARAMETER_TASK_INSTANCE_ID, Integer.toString(taskExecutionContext.getTaskInstanceId()));
if (globalParams != null && localParams != null) {
if (varParams.size() != 0) {
globalParams.putAll(varParams);
}
if (localParams.size() != 0) {
globalParams.putAll(localParams);
} else if (globalParams == null && localParams != null) {
globalParams = localParams;
}
if (varParams != null) {
varParams.putAll(globalParams);
globalParams = varParams;
}
Iterator<Map.Entry<String, Property>> iter = globalParams.entrySet().iterator();
while (iter.hasNext()) {
Map.Entry<String, Property> en = iter.next();
@ -143,16 +142,15 @@ public class ParamUtils {
* @return parameters map
*/
public static Map<String, Property> getUserDefParamsMap(Map<String, String> definedParams) {
Map<String, Property> userDefParamsMaps = new HashMap<>();
if (definedParams != null) {
Map<String, Property> userDefParamsMaps = new HashMap<>();
Iterator<Map.Entry<String, String>> iter = definedParams.entrySet().iterator();
while (iter.hasNext()) {
Map.Entry<String, String> en = iter.next();
Property property = new Property(en.getKey(), Direct.IN, DataType.VARCHAR, en.getValue());
userDefParamsMaps.put(property.getProp(),property);
}
return userDefParamsMaps;
}
return null;
return userDefParamsMaps;
}
}

View File

@ -243,7 +243,11 @@ public class SqlTask extends AbstractTaskExecutor {
int rowCount = 0;
int limit = sqlParameters.getLimit() == 0 ? QUERY_LIMIT : sqlParameters.getLimit();
while (rowCount < limit && resultSet.next()) {
while (resultSet.next()) {
if (rowCount == limit) {
logger.info("sql result limit : {} exceeding results are filtered", limit);
break;
}
ObjectNode mapOfColValues = JSONUtils.createObjectNode();
for (int i = 1; i <= num; i++) {
mapOfColValues.set(md.getColumnLabel(i), JSONUtils.toJsonNode(resultSet.getObject(i)));
@ -252,17 +256,12 @@ public class SqlTask extends AbstractTaskExecutor {
rowCount++;
}
int displayRows = sqlParameters.getDisplayRows() > 0 ? sqlParameters.getDisplayRows() : TaskConstants.DEFAULT_DISPLAY_ROWS;
displayRows = Math.min(displayRows, resultJSONArray.size());
displayRows = Math.min(displayRows, rowCount);
logger.info("display sql result {} rows as follows:", displayRows);
for (int i = 0; i < displayRows; i++) {
String row = JSONUtils.toJsonString(resultJSONArray.get(i));
logger.info("row {} : {}", i + 1, row);
}
if (resultSet.next()) {
logger.info("sql result limit : {} exceeding results are filtered", limit);
String log = String.format("sql result limit : %d exceeding results are filtered", limit);
resultJSONArray.add(JSONUtils.toJsonNode(log));
}
}
String result = JSONUtils.toJsonString(resultJSONArray);
if (sqlParameters.getSendEmail() == null || sqlParameters.getSendEmail()) {

View File

@ -30,7 +30,7 @@
<properties>
<node.version>v16.13.1</node.version>
<npm.version>8.1.2</npm.version>
<pnpm.version>v6.32.6</pnpm.version>
<sonar.sources>src</sonar.sources>
</properties>
@ -45,19 +45,19 @@
<version>${frontend-maven-plugin.version}</version>
<executions>
<execution>
<id>install node and npm</id>
<id>install node and pnpm</id>
<goals>
<goal>install-node-and-npm</goal>
<goal>install-node-and-pnpm</goal>
</goals>
<configuration>
<nodeVersion>${node.version}</nodeVersion>
<npmVersion>${npm.version}</npmVersion>
<pnpmVersion>${pnpm.version}</pnpmVersion>
</configuration>
</execution>
<execution>
<id>npm install</id>
<id>pnpm install</id>
<goals>
<goal>npm</goal>
<goal>pnpm</goal>
</goals>
<phase>generate-resources</phase>
<configuration>
@ -65,9 +65,9 @@
</configuration>
</execution>
<execution>
<id>npm run build:prod</id>
<id>pnpm run build:prod</id>
<goals>
<goal>npm</goal>
<goal>pnpm</goal>
</goals>
<configuration>
<arguments>run build:prod</arguments>

View File

@ -38,7 +38,7 @@ function initChart<Opt extends ECBasicOption>(
const init = () => {
chart = globalProperties?.echarts.init(
domRef.value,
themeStore.darkTheme && 'dark-bold'
themeStore.darkTheme ? 'dark-bold' : 'macarons'
)
chart && chart.setOption(option)
}

View File

@ -1113,7 +1113,7 @@ const security = {
Secret: 'Secret',
users: 'Users',
userSendMsg: 'UserSendMsg',
agentId: 'AgentId',
'agentId/chatId': 'AgentId or ChatId',
showType: 'Show Type',
receivers: 'Receivers',
receiverCcs: 'ReceiverCcs',

View File

@ -1098,7 +1098,7 @@ const security = {
Secret: '密钥',
users: '群员',
userSendMsg: '群员信息',
agentId: '应用ID',
'agentId/chatId': '应用ID或群聊ID',
showType: '内容展示类型',
receivers: '收件人',
receiverCcs: '抄送人',

View File

@ -93,7 +93,7 @@
<spotbugs.version>3.1.12</spotbugs.version>
<checkstyle.version>3.1.2</checkstyle.version>
<curator.test>2.12.0</curator.test>
<frontend-maven-plugin.version>1.6</frontend-maven-plugin.version>
<frontend-maven-plugin.version>1.12.1</frontend-maven-plugin.version>
<maven-compiler-plugin.version>3.3</maven-compiler-plugin.version>
<maven-assembly-plugin.version>3.3.0</maven-assembly-plugin.version>
<maven-release-plugin.version>2.5.3</maven-release-plugin.version>

View File

@ -16,7 +16,7 @@
# limitations under the License.
#
usage="Usage: dolphinscheduler-daemon.sh (start|stop|status) <api-server|master-server|worker-server|alert-server|python-gateway-server|standalone-server> "
usage="Usage: dolphinscheduler-daemon.sh (start|stop|status) <api-server|master-server|worker-server|alert-server|standalone-server> "
# if no args specified, show usage
if [ $# -le 1 ]; then
@ -61,8 +61,6 @@ elif [ "$command" = "alert-server" ]; then
log=$DOLPHINSCHEDULER_HOME/alert-server/logs/$command-$HOSTNAME.out
elif [ "$command" = "standalone-server" ]; then
log=$DOLPHINSCHEDULER_HOME/standalone-server/logs/$command-$HOSTNAME.out
elif [ "$command" = "python-gateway-server" ]; then
log=$DOLPHINSCHEDULER_HOME/python-gateway-server/logs/$command-$HOSTNAME.out
else
echo "Error: No command named '$command' was found."
exit 1

View File

@ -48,11 +48,6 @@ alertServer=${alertServer:-"ds3"}
# Example for hostname: apiServers="ds1", Example for IP: apiServers="192.168.8.1"
apiServers=${apiServers:-"ds1"}
# A comma separated list of machine hostname or IP would be installed Python gateway server, it
# must be a subset of configuration `ips`.
# Example for hostname: pythonGatewayServers="ds1", Example for IP: pythonGatewayServers="192.168.8.1"
pythonGatewayServers=${pythonGatewayServers:-"ds1"}
# The directory to install DolphinScheduler for all machine we config above. It will automatically be created by `install.sh` script if not exists.
# Do not set this configuration same as the current path (pwd)
installPath=${installPath:-"/tmp/dolphinscheduler"}

View File

@ -49,7 +49,7 @@ do
echo "scp dirs to $host/$installPath starting"
ssh -p $sshPort $host "cd $installPath/; rm -rf bin/ conf/ lib/ script/ sql/ ui/"
for dsDir in bin master-server worker-server alert-server api-server ui python-gateway-server
for dsDir in bin master-server worker-server alert-server api-server ui
do
# if worker in workersGroupMap
if [[ "${workersGroupMap[${host}]}" ]]; then

View File

@ -56,13 +56,6 @@ do
ssh -p $sshPort $apiServer "cd $installPath/; sh bin/dolphinscheduler-daemon.sh start api-server;"
done
pythonGatewayHost=(${pythonGatewayServers//,/ })
for pythonGatewayServer in "${pythonGatewayHost[@]}"
do
echo "$pythonGatewayServer python gateway server is starting"
ssh -p $sshPort $pythonGatewayServer "cd $installPath/; sh bin/dolphinscheduler-daemon.sh start python-gateway-server;"
done
# query server status
echo "query server status"
cd $installPath/; sh bin/status-all.sh

View File

@ -74,11 +74,3 @@ do
apiState=`ssh -p $sshPort $apiServer "cd $installPath/; sh bin/dolphinscheduler-daemon.sh status api-server;"`
echo "$apiServer $apiState"
done
# python gateway server check state
pythonGatewayHost=(${pythonGatewayServers//,/ })
for pythonGatewayServer in "${pythonGatewayHost[@]}"
do
pythonGatewayState=`ssh -p $sshPort $pythonGatewayServer "cd $installPath/; sh bin/dolphinscheduler-daemon.sh status python-gateway-server;"`
echo "$pythonGatewayServer $pythonGatewayState"
done

View File

@ -54,10 +54,3 @@ do
echo "$apiServer api server is stopping"
ssh -p $sshPort $apiServer "cd $installPath/; sh bin/dolphinscheduler-daemon.sh stop api-server;"
done
pythonGatewayHost=(${pythonGatewayServers//,/ })
for pythonGatewayServer in "${pythonGatewayHost[@]}"
do
echo "$pythonGatewayServer python gateway server is stopping"
ssh -p $sshPort $pythonGatewayServer "cd $installPath/; sh bin/dolphinscheduler-daemon.sh stop python-gateway-server;"
done