You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dolphinscheduler.apache.org by zh...@apache.org on 2022/07/28 03:14:05 UTC
[dolphinscheduler] branch 3.0.0-prepare updated (cec5e88319 -> 88712b423a)
This is an automated email from the ASF dual-hosted git repository.
zhongjiajie pushed a change to branch 3.0.0-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
from cec5e88319 [Bug-11071] Fix task local params NPE (#11166)
new d3f28c8411 [python] Fix submit and run error (#10792)
new b2ad6f3f8e [python] Integrate test run strategy change (#10825)
new c67f9ee85d [fix] update queue and tenant with same code error (#10877)
new e4ab488426 [python] Fix tasks with multiple upstream and workflow query error (#10941)
new 4da1240b0e [Python] Supports creating or editing resources. (#10823)
new 88712b423a [fix] Some cherry pick error
The 6 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.github/workflows/py-ci.yml | 85 +++++---
.../apache/dolphinscheduler/api/enums/Status.java | 2 +-
.../api/exceptions/ApiExceptionHandler.java | 6 +
.../api/exceptions/ServiceException.java | 7 +
.../dolphinscheduler/api/python/PythonGateway.java | 61 ++----
.../dolphinscheduler/api/service/QueueService.java | 10 +-
.../api/service/ResourcesService.java | 24 ++
.../api/service/TenantService.java | 18 +-
.../api/service/impl/QueueServiceImpl.java | 197 +++++++----------
.../api/service/impl/ResourcesServiceImpl.java | 122 ++++++++++-
.../api/service/impl/TenantServiceImpl.java | 176 ++++++++-------
.../api/controller/QueueControllerTest.java | 60 +++--
.../api/python/PythonGatewayTest.java | 22 +-
.../api/service/QueueServiceTest.java | 166 ++++++++------
.../api/service/ResourcesServiceTest.java | 60 ++++-
.../api/service/TenantServiceTest.java | 242 +++++++++++++--------
.../apache/dolphinscheduler/common/Constants.java | 5 +
.../apache/dolphinscheduler/dao/entity/Queue.java | 20 ++
.../apache/dolphinscheduler/dao/entity/Tenant.java | 21 ++
.../dolphinscheduler/dao/mapper/QueueMapper.java | 7 +-
.../dolphinscheduler/dao/mapper/QueueMapper.xml | 12 +-
.../dao/mapper/TaskDefinitionMapper.xml | 1 +
.../pydolphinscheduler/core/process_definition.py | 15 +-
.../core/{base_side.py => resource.py} | 31 +--
.../tests/core/test_process_definition.py | 9 +-
.../core/test_resource_definition.py} | 28 ++-
.../tests/integration/test_submit_examples.py | 22 +-
27 files changed, 926 insertions(+), 503 deletions(-)
copy dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/{base_side.py => resource.py} (62%)
copy dolphinscheduler-python/pydolphinscheduler/{src/pydolphinscheduler/side/worker_group.py => tests/core/test_resource_definition.py} (59%)
[dolphinscheduler] 03/06: [fix] update queue and tenant with same code error (#10877)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhongjiajie pushed a commit to branch 3.0.0-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
commit c67f9ee85d67d655a9a7abc5d1730de7ba270123
Author: Jiajie Zhong <zh...@gmail.com>
AuthorDate: Mon Jul 11 11:06:30 2022 +0800
[fix] update queue and tenant with same code error (#10877)
in ae6aa53f I use the same logic test to create or update
tenant and queue which failed create with the same value,
this patch will fix it. and also I use the constructor to reduce
the new object logic here
(cherry picked from commit a6154220dc9fd78a894f9fc47c78378cba1d03cd)
---
.../api/service/impl/QueueServiceImpl.java | 113 ++++++++++-----------
.../api/service/impl/TenantServiceImpl.java | 92 ++++++++---------
.../api/service/QueueServiceTest.java | 11 ++
.../api/service/TenantServiceTest.java | 6 ++
.../apache/dolphinscheduler/dao/entity/Queue.java | 20 ++++
.../apache/dolphinscheduler/dao/entity/Tenant.java | 21 ++++
6 files changed, 154 insertions(+), 109 deletions(-)
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
index 317bcd2cee..64ff494342 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
@@ -17,9 +17,9 @@
package org.apache.dolphinscheduler.api.service.impl;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
-import org.apache.commons.lang3.StringUtils;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_CREATE;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_UPDATE;
+
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.QueueService;
@@ -33,9 +33,10 @@ import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.QueueMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
+import org.apache.commons.lang3.StringUtils;
+
import java.util.ArrayList;
import java.util.Collections;
-import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
@@ -49,8 +50,8 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
-import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_CREATE;
-import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_UPDATE;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
/**
* queue service impl
@@ -67,41 +68,45 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
private UserMapper userMapper;
/**
- * Valid both queue and queueName when we want to create or update queue object
+ * Check the queue new object valid or not
*
- * @param queue queue value
- * @param queueName queue name
+ * @param queue The queue object want to create
*/
- private void queueValid(String queue, String queueName) throws ServiceException {
- if (StringUtils.isEmpty(queue)) {
+ private void createQueueValid(Queue queue) throws ServiceException {
+ if (StringUtils.isEmpty(queue.getQueue())) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
- } else if (StringUtils.isEmpty(queueName)) {
+ } else if (StringUtils.isEmpty(queue.getQueueName())) {
throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
- } else if (checkQueueExist(queue)) {
- throw new ServiceException(Status.QUEUE_VALUE_EXIST, queue);
- } else if (checkQueueNameExist(queueName)) {
- throw new ServiceException(Status.QUEUE_NAME_EXIST, queueName);
+ } else if (checkQueueExist(queue.getQueue())) {
+ throw new ServiceException(Status.QUEUE_VALUE_EXIST, queue.getQueue());
+ } else if (checkQueueNameExist(queue.getQueueName())) {
+ throw new ServiceException(Status.QUEUE_NAME_EXIST, queue.getQueueName());
}
}
/**
- * Insert one single new Queue record to database
+ * Check queue update object valid or not
*
- * @param queue queue value
- * @param queueName queue name
- * @return Queue
+ * @param existsQueue The exists queue object
+ * @param updateQueue The queue object want to update
*/
- private Queue createObjToDB(String queue, String queueName) {
- Queue queueObj = new Queue();
- Date now = new Date();
-
- queueObj.setQueue(queue);
- queueObj.setQueueName(queueName);
- queueObj.setCreateTime(now);
- queueObj.setUpdateTime(now);
- // save
- queueMapper.insert(queueObj);
- return queueObj;
+ private void updateQueueValid(Queue existsQueue, Queue updateQueue) throws ServiceException {
+ // Check the exists queue and the necessary of update operation, in not exist checker have to use updateQueue to avoid NPE
+ if (Objects.isNull(existsQueue)) {
+ throw new ServiceException(Status.QUEUE_NOT_EXIST, updateQueue.getQueue());
+ } else if (Objects.equals(existsQueue, updateQueue)) {
+ throw new ServiceException(Status.NEED_NOT_UPDATE_QUEUE);
+ }
+ // Check the update queue parameters
+ else if (StringUtils.isEmpty(updateQueue.getQueue())) {
+ throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
+ } else if (StringUtils.isEmpty(updateQueue.getQueueName())) {
+ throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
+ } else if (!Objects.equals(updateQueue.getQueue(), existsQueue.getQueue()) && checkQueueExist(updateQueue.getQueue())) {
+ throw new ServiceException(Status.QUEUE_VALUE_EXIST, updateQueue.getQueue());
+ } else if (!Objects.equals(updateQueue.getQueueName(), existsQueue.getQueueName()) && checkQueueNameExist(updateQueue.getQueueName())) {
+ throw new ServiceException(Status.QUEUE_NAME_EXIST, updateQueue.getQueueName());
+ }
}
/**
@@ -169,11 +174,14 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
if (isNotAdmin(loginUser, result)) {
return result;
}
- queueValid(queue, queueName);
- Queue newQueue = createObjToDB(queue, queueName);
- result.put(Constants.DATA_LIST, newQueue);
+ Queue queueObj = new Queue(queueName, queue);
+ createQueueValid(queueObj);
+ queueMapper.insert(queueObj);
+
+ result.put(Constants.DATA_LIST, queueObj);
putMsg(result, Status.SUCCESS);
+ permissionPostHandle(AuthorizationType.QUEUE, loginUser.getId(), Collections.singletonList(queueObj.getId()), logger);
return result;
}
@@ -193,35 +201,20 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
return result;
}
- queueValid(queue, queueName);
-
- Queue queueObj = queueMapper.selectById(id);
- if (Objects.isNull(queueObj)) {
- throw new ServiceException(Status.QUEUE_NOT_EXIST, queue);
- }
-
- // whether queue value or queueName is changed
- if (queue.equals(queueObj.getQueue()) && queueName.equals(queueObj.getQueueName())) {
- throw new ServiceException(Status.NEED_NOT_UPDATE_QUEUE);
- }
+ Queue updateQueue = new Queue(id, queueName, queue);
+ Queue existsQueue = queueMapper.selectById(id);
+ updateQueueValid(existsQueue, updateQueue);
// check old queue using by any user
- if (checkIfQueueIsInUsing(queueObj.getQueueName(), queueName)) {
+ if (checkIfQueueIsInUsing(existsQueue.getQueueName(), updateQueue.getQueueName())) {
//update user related old queue
- Integer relatedUserNums = userMapper.updateUserQueue(queueObj.getQueueName(), queueName);
+ Integer relatedUserNums = userMapper.updateUserQueue(existsQueue.getQueueName(), updateQueue.getQueueName());
logger.info("old queue have related {} user, exec update user success.", relatedUserNums);
}
- // update queue
- Date now = new Date();
- queueObj.setQueue(queue);
- queueObj.setQueueName(queueName);
- queueObj.setUpdateTime(now);
-
- queueMapper.updateById(queueObj);
+ queueMapper.updateById(updateQueue);
putMsg(result, Status.SUCCESS);
-
return result;
}
@@ -235,7 +228,9 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
@Override
public Result<Object> verifyQueue(String queue, String queueName) {
Result<Object> result = new Result<>();
- queueValid(queue, queueName);
+
+ Queue queueValidator = new Queue(queueName, queue);
+ createQueueValid(queueValidator);
putMsg(result, Status.SUCCESS);
return result;
@@ -287,10 +282,12 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
*/
@Override
public Queue createQueueIfNotExists(String queue, String queueName) {
- queueValid(queue, queueName);
+ Queue queueObj = new Queue(queueName, queue);
+ createQueueValid(queueObj);
Queue existsQueue = queueMapper.queryQueueName(queue, queueName);
if (Objects.isNull(existsQueue)) {
- return createObjToDB(queue, queueName);
+ queueMapper.insert(queueObj);
+ return queueObj;
}
return existsQueue;
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
index c4cca300fb..6e7cfecb83 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
@@ -86,44 +86,43 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
private StorageOperate storageOperate;
/**
- * Valid tenantCode when we want to create or update tenant object
+ * Check the tenant new object valid or not
*
- * @param tenantCode Tenant code of tenant object
- * @return Optional of Status map
+ * @param tenant The tenant object want to create
*/
- private void tenantCodeValid(String tenantCode) throws ServiceException {
- Map<String, Object> result = new HashMap<>();
- if (StringUtils.isEmpty(tenantCode)) {
- throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, tenantCode);
- } else if (StringUtils.length(tenantCode) > TENANT_FULL_NAME_MAX_LENGTH) {
+ private void createTenantValid(Tenant tenant) throws ServiceException {
+ if (StringUtils.isEmpty(tenant.getTenantCode())) {
+ throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, tenant.getTenantCode());
+ } else if (StringUtils.length(tenant.getTenantCode()) > TENANT_FULL_NAME_MAX_LENGTH) {
throw new ServiceException(Status.TENANT_FULL_NAME_TOO_LONG_ERROR);
- } else if (!RegexUtils.isValidLinuxUserName(tenantCode)) {
+ } else if (!RegexUtils.isValidLinuxUserName(tenant.getTenantCode())) {
throw new ServiceException(Status.CHECK_OS_TENANT_CODE_ERROR);
- } else if (checkTenantExists(tenantCode)) {
- throw new ServiceException(Status.OS_TENANT_CODE_EXIST, tenantCode);
+ } else if (checkTenantExists(tenant.getTenantCode())) {
+ throw new ServiceException(Status.OS_TENANT_CODE_EXIST, tenant.getTenantCode());
}
}
/**
- * Insert one single new Tenant record to database
+ * Check tenant update object valid or not
*
- * @param tenantCode new Tenant object tenant code
- * @param desc new Tenant object description
- * @param queueId The Queue id of new Tenant object
- * @return Tenant
+ * @param existsTenant The exists queue object
+ * @param updateTenant The queue object want to update
*/
- private Tenant createObjToDB(String tenantCode, String desc, int queueId) {
- Tenant tenant = new Tenant();
- Date now = new Date();
-
- tenant.setTenantCode(tenantCode);
- tenant.setQueueId(queueId);
- tenant.setDescription(desc);
- tenant.setCreateTime(now);
- tenant.setUpdateTime(now);
- // save
- tenantMapper.insert(tenant);
- return tenant;
+ private void updateTenantValid(Tenant existsTenant, Tenant updateTenant) throws ServiceException {
+ // Check the exists tenant
+ if (Objects.isNull(existsTenant)) {
+ throw new ServiceException(Status.TENANT_NOT_EXIST);
+ }
+ // Check the update tenant parameters
+ else if (StringUtils.isEmpty(updateTenant.getTenantCode())) {
+ throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, updateTenant.getTenantCode());
+ } else if (StringUtils.length(updateTenant.getTenantCode()) > TENANT_FULL_NAME_MAX_LENGTH) {
+ throw new ServiceException(Status.TENANT_FULL_NAME_TOO_LONG_ERROR);
+ } else if (!RegexUtils.isValidLinuxUserName(updateTenant.getTenantCode())) {
+ throw new ServiceException(Status.CHECK_OS_TENANT_CODE_ERROR);
+ } else if (!Objects.equals(existsTenant.getTenantCode(), updateTenant.getTenantCode()) && checkTenantExists(updateTenant.getTenantCode())) {
+ throw new ServiceException(Status.OS_TENANT_CODE_EXIST, updateTenant.getTenantCode());
+ }
}
/**
@@ -148,15 +147,16 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
- tenantCodeValid(tenantCode);
+ Tenant tenant = new Tenant(tenantCode, desc, queueId);
+ createTenantValid(tenant);
+ tenantMapper.insert(tenant);
- Tenant newTenant = createObjToDB(tenantCode, desc, queueId);
// if storage startup
if (PropertyUtils.getResUploadStartupState()) {
storageOperate.createTenantDirIfNotExists(tenantCode);
}
- permissionPostHandle(AuthorizationType.TENANT, loginUser.getId(), Collections.singletonList(newTenant.getId()), logger);
- result.put(Constants.DATA_LIST, newTenant);
+ permissionPostHandle(AuthorizationType.TENANT, loginUser.getId(), Collections.singletonList(tenant.getId()), logger);
+ result.put(Constants.DATA_LIST, tenant);
putMsg(result, Status.SUCCESS);
return result;
}
@@ -212,30 +212,18 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
- Tenant tenant = tenantMapper.queryById(id);
-
- if (Objects.isNull(tenant)) {
- throw new ServiceException(Status.TENANT_NOT_EXIST);
- }
-
- tenantCodeValid(tenantCode);
+ Tenant updateTenant = new Tenant(id, tenantCode, desc, queueId);
+ Tenant existsTenant = tenantMapper.queryById(id);
+ updateTenantValid(existsTenant, updateTenant);
// updateProcessInstance tenant
/**
* if the tenant code is modified, the original resource needs to be copied to the new tenant.
*/
- if (!tenant.getTenantCode().equals(tenantCode) && PropertyUtils.getResUploadStartupState()) {
+ if (!Objects.equals(existsTenant.getTenantCode(), updateTenant.getTenantCode()) && PropertyUtils.getResUploadStartupState()) {
storageOperate.createTenantDirIfNotExists(tenantCode);
}
-
- Date now = new Date();
-
- if (queueId != 0) {
- tenant.setQueueId(queueId);
- }
- tenant.setDescription(desc);
- tenant.setUpdateTime(now);
- tenantMapper.updateById(tenant);
+ tenantMapper.updateById(updateTenant);
putMsg(result, Status.SUCCESS);
return result;
@@ -378,8 +366,10 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
return tenantMapper.queryByTenantCode(tenantCode);
}
- tenantCodeValid(tenantCode);
Queue newQueue = queueService.createQueueIfNotExists(queue, queueName);
- return createObjToDB(tenantCode, desc, newQueue.getId());
+ Tenant tenant = new Tenant(tenantCode, desc, newQueue.getId());
+ createTenantValid(tenant);
+ tenantMapper.insert(tenant);
+ return tenant;
}
}
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
index b6f1fc6e3f..f32723225c 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
@@ -82,6 +82,7 @@ public class QueueServiceTest {
private static final String QUEUE_NAME = "queueName";
private static final String EXISTS = "exists";
private static final String NOT_EXISTS = "not_exists";
+ private static final String NOT_EXISTS_FINAL = "not_exists_final";
@Before
public void setUp() {
@@ -170,6 +171,16 @@ public class QueueServiceTest {
Mockito.when(userMapper.existUser(Mockito.anyString())).thenReturn(false);
Map<String, Object> result = queueService.updateQueue(getLoginUser(), 1, NOT_EXISTS, NOT_EXISTS);
Assert.assertEquals(Status.SUCCESS.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
+
+ // success update with same queue name
+ Mockito.when(queueMapper.existQueue(NOT_EXISTS_FINAL, null)).thenReturn(false);
+ result = queueService.updateQueue(getLoginUser(), 1, NOT_EXISTS_FINAL, NOT_EXISTS);
+ Assert.assertEquals(Status.SUCCESS.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
+
+ // success update with same queue value
+ Mockito.when(queueMapper.existQueue(null, NOT_EXISTS_FINAL)).thenReturn(false);
+ result = queueService.updateQueue(getLoginUser(), 1, NOT_EXISTS, NOT_EXISTS_FINAL);
+ Assert.assertEquals(Status.SUCCESS.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
}
@Test
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
index 9ce34afe7a..0229db3c02 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
@@ -166,11 +166,17 @@ public class TenantServiceTest {
Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.TENANT, getLoginUser().getId(), TENANT_UPDATE, baseServiceLogger)).thenReturn(true);
Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.TENANT, null, 0, baseServiceLogger)).thenReturn(true);
+ // update not exists tenant
Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.updateTenant(getLoginUser(), 912222, tenantCode, 1, tenantDesc));
Assertions.assertEquals(Status.TENANT_NOT_EXIST.getMsg(), exception.getMessage());
+ // success
Map<String, Object> result = tenantService.updateTenant(getLoginUser(), 1, tenantCode, 1, tenantDesc);
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+
+ // success update with same tenant code
+ result = tenantService.updateTenant(getLoginUser(), 1, tenantCode, 1, tenantDesc);
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@Test
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Queue.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Queue.java
index 57b678d2a7..cc423e13ca 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Queue.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Queue.java
@@ -51,6 +51,26 @@ public class Queue {
*/
private Date updateTime;
+ public Queue() {
+ }
+
+ public Queue(String queueName, String queue) {
+ Date now = new Date();
+ this.queueName = queueName;
+ this.queue = queue;
+ this.createTime = now;
+ this.updateTime = now;
+ }
+
+ public Queue(int id, String queueName, String queue) {
+ Date now = new Date();
+ this.id = id;
+ this.queueName = queueName;
+ this.queue = queue;
+ this.createTime = now;
+ this.updateTime = now;
+ }
+
public int getId() {
return id;
}
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Tenant.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Tenant.java
index 1b0896b1d8..d6a548151e 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Tenant.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/entity/Tenant.java
@@ -72,6 +72,27 @@ public class Tenant {
*/
private Date updateTime;
+ public Tenant() {
+ }
+
+ public Tenant(String tenantCode, String description, int queueId) {
+ Date now = new Date();
+ this.tenantCode = tenantCode;
+ this.description = description;
+ this.queueId = queueId;
+ this.createTime = now;
+ this.updateTime = now;
+ }
+
+ public Tenant(int id, String tenantCode, String description, int queueId) {
+ Date now = new Date();
+ this.id = id;
+ this.tenantCode = tenantCode;
+ this.description = description;
+ this.queueId = queueId;
+ this.createTime = now;
+ this.updateTime = now;
+ }
public int getId() {
return id;
[dolphinscheduler] 04/06: [python] Fix tasks with multiple upstream and workflow query error (#10941)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhongjiajie pushed a commit to branch 3.0.0-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
commit e4ab488426ab92f48f7eb1c525f4e1e050cdd27d
Author: Jiajie Zhong <zh...@gmail.com>
AuthorDate: Thu Jul 14 22:06:07 2022 +0800
[python] Fix tasks with multiple upstream and workflow query error (#10941)
* when task with more than one upstreams, mapper
TaskDefinitionMapper method queryByName will return
more than one record, and failed the mybatis result
type, so we have to add `limit 1` to it to
* add multiple runs of example in integrate test
* Change from subprocess.Popen to subprocess.call_check
in integrating test which will raise an error when failed
(cherry picked from commit 81930e54208f7c0f305c0b9f8846149bfc38f428)
---
.github/workflows/py-ci.yml | 4 +--
.../dolphinscheduler/api/python/PythonGateway.java | 7 +++---
.../api/service/impl/QueueServiceImpl.java | 12 ++++-----
.../api/service/impl/TenantServiceImpl.java | 4 +--
.../api/service/QueueServiceTest.java | 15 +++++++++++
.../api/service/TenantServiceTest.java | 29 ++++++++++++++++++++++
.../dao/mapper/TaskDefinitionMapper.xml | 1 +
.../pydolphinscheduler/core/process_definition.py | 14 +++++++++++
.../tests/integration/test_submit_examples.py | 22 ++++++++++++----
9 files changed, 90 insertions(+), 18 deletions(-)
diff --git a/.github/workflows/py-ci.yml b/.github/workflows/py-ci.yml
index eddd65b6da..2c8e578a7a 100644
--- a/.github/workflows/py-ci.yml
+++ b/.github/workflows/py-ci.yml
@@ -51,7 +51,7 @@ jobs:
not-docs:
- '!(docs/**)'
py-change:
- - 'dolphinscheduler-python/pydolphinscheduler'
+ - 'dolphinscheduler-python/pydolphinscheduler/**'
lint:
name: Lint
if: ${{ (needs.paths-filter.outputs.py-change == 'true') || (github.event_name == 'push') }}
@@ -163,7 +163,7 @@ jobs:
- name: Install Dependences
run: |
python -m pip install --upgrade ${{ env.DEPENDENCES }}
- - name: Run Tests Build Docs
+ - name: Run Integrate Tests
run: |
python -m tox -vv -e integrate-test
result:
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
index 71db35cb2a..89b396ae98 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
@@ -174,10 +174,11 @@ public class PythonGateway {
}
ProcessDefinition processDefinition = processDefinitionMapper.queryByDefineName(project.getCode(), processDefinitionName);
+ // In the case project exists, but current process definition still not created, we should also return the init version of it
if (processDefinition == null) {
- String msg = String.format("Can not find valid process definition by name %s", processDefinitionName);
- logger.error(msg);
- throw new IllegalArgumentException(msg);
+ result.put("code", CodeGenerateUtils.getInstance().genCode());
+ result.put("version", 0L);
+ return result;
}
TaskDefinition taskDefinition = taskDefinitionMapper.queryByName(project.getCode(), processDefinition.getCode(), taskName);
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
index 64ff494342..bce236250e 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
@@ -282,14 +282,14 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
*/
@Override
public Queue createQueueIfNotExists(String queue, String queueName) {
- Queue queueObj = new Queue(queueName, queue);
- createQueueValid(queueObj);
Queue existsQueue = queueMapper.queryQueueName(queue, queueName);
- if (Objects.isNull(existsQueue)) {
- queueMapper.insert(queueObj);
- return queueObj;
+ if (!Objects.isNull(existsQueue)) {
+ return existsQueue;
}
- return existsQueue;
+ Queue queueObj = new Queue(queueName, queue);
+ createQueueValid(queueObj);
+ queueMapper.insert(queueObj);
+ return queueObj;
}
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
index 6e7cfecb83..f07966af76 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
@@ -366,8 +366,8 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
return tenantMapper.queryByTenantCode(tenantCode);
}
- Queue newQueue = queueService.createQueueIfNotExists(queue, queueName);
- Tenant tenant = new Tenant(tenantCode, desc, newQueue.getId());
+ Queue queueObj = queueService.createQueueIfNotExists(queue, queueName);
+ Tenant tenant = new Tenant(tenantCode, desc, queueObj.getId());
createTenantValid(tenant);
tenantMapper.insert(tenant);
return tenant;
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
index f32723225c..cd8cdacce2 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
@@ -212,6 +212,21 @@ public class QueueServiceTest {
Assert.assertEquals(result.getCode().intValue(), Status.SUCCESS.getCode());
}
+ @Test
+ public void testCreateQueueIfNotExists() {
+ Queue queue;
+
+ // queue exists
+ Mockito.when(queueMapper.queryQueueName(QUEUE, QUEUE_NAME)).thenReturn(getQUEUE());
+ queue = queueService.createQueueIfNotExists(QUEUE, QUEUE_NAME);
+ Assert.assertEquals(getQUEUE(), queue);
+
+ // queue not exists
+ Mockito.when(queueMapper.queryQueueName(QUEUE, QUEUE_NAME)).thenReturn(null);
+ queue = queueService.createQueueIfNotExists(QUEUE, QUEUE_NAME);
+ Assert.assertEquals(new Queue(QUEUE_NAME, QUEUE), queue);
+ }
+
/**
* create admin user
*/
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
index 0229db3c02..62fa91d2c5 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
@@ -34,6 +34,7 @@ import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.entity.Queue;
import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
@@ -77,6 +78,9 @@ public class TenantServiceTest {
@InjectMocks
private TenantServiceImpl tenantService;
+ @Mock
+ private QueueService queueService;
+
@Mock
private TenantMapper tenantMapper;
@@ -94,6 +98,8 @@ public class TenantServiceTest {
private static final String tenantCode = "hayden";
private static final String tenantDesc = "This is the tenant desc";
+ private static final String queue = "queue";
+ private static final String queueName = "queue_name";
@Test
public void testCreateTenant() throws Exception {
@@ -229,6 +235,23 @@ public class TenantServiceTest {
Assert.assertEquals(Status.SUCCESS.getMsg(), result.getMsg());
}
+ @Test
+ public void testCreateTenantIfNotExists() {
+ Tenant tenant;
+
+ // Tenant exists
+ Mockito.when(tenantMapper.existTenant(tenantCode)).thenReturn(true);
+ Mockito.when(tenantMapper.queryByTenantCode(tenantCode)).thenReturn(getTenant());
+ tenant = tenantService.createTenantIfNotExists(tenantCode, tenantDesc, queue, queueName);
+ Assert.assertEquals(getTenant(), tenant);
+
+ // Tenant not exists
+ Mockito.when(tenantMapper.existTenant(tenantCode)).thenReturn(false);
+ Mockito.when(queueService.createQueueIfNotExists(queue, queueName)).thenReturn(getQueue());
+ tenant = tenantService.createTenantIfNotExists(tenantCode, tenantDesc, queue, queueName);
+ Assert.assertEquals(new Tenant(tenantCode, tenantDesc, getQueue().getId()), tenant);
+ }
+
/**
* get user
*/
@@ -284,4 +307,10 @@ public class TenantServiceTest {
return processDefinitions;
}
+ private Queue getQueue() {
+ Queue queue = new Queue();
+ queue.setId(1);
+ return queue;
+ }
+
}
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.xml
index d7cf5f11bd..6da42c6714 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/TaskDefinitionMapper.xml
@@ -41,6 +41,7 @@
and td.name = #{name}
and ptr.process_definition_code = #{processCode}
and td.code = ptr.post_task_code
+ limit 1
</select>
<select id="queryAllDefinitionList" resultType="org.apache.dolphinscheduler.dao.entity.TaskDefinition">
select
diff --git a/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/process_definition.py b/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/process_definition.py
index 5f34b6db60..dd18dafa0f 100644
--- a/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/process_definition.py
+++ b/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/process_definition.py
@@ -63,6 +63,9 @@ class ProcessDefinition(Base):
thought Web UI after it :func:`submit` or :func:`run`. It will create a new project belongs to
``user`` if it does not exists. And when ``project`` exists but project's create do not belongs
to ``user``, will grant `project` to ``user`` automatically.
+ :param resource_list: Resource files required by the current process definition.You can create and modify
+ resource files from this field. When the process definition is submitted, these resource files are
+ also submitted along with it.
"""
# key attribute for identify ProcessDefinition object
@@ -88,6 +91,7 @@ class ProcessDefinition(Base):
"tasks",
"task_definition_json",
"task_relation_json",
+ "resource_list",
}
def __init__(
@@ -107,6 +111,7 @@ class ProcessDefinition(Base):
timeout: Optional[int] = 0,
release_state: Optional[str] = configuration.WORKFLOW_RELEASE_STATE,
param: Optional[Dict] = None,
+ resource_list: Optional[List] = None,
):
super().__init__(name, description)
self.schedule = schedule
@@ -132,6 +137,7 @@ class ProcessDefinition(Base):
# TODO how to fix circle import
self._task_relations: set["TaskRelation"] = set() # noqa: F821
self._process_definition_code = None
+ self.resource_list = resource_list or []
def __enter__(self) -> "ProcessDefinition":
ProcessDefinitionContext.set(self)
@@ -407,6 +413,14 @@ class ProcessDefinition(Base):
None,
None,
)
+ if len(self.resource_list) > 0:
+ for res in self.resource_list:
+ gateway.entry_point.createOrUpdateResource(
+ self._user,
+ res.name,
+ res.description,
+ res.content,
+ )
return self._process_definition_code
def start(self) -> None:
diff --git a/dolphinscheduler-python/pydolphinscheduler/tests/integration/test_submit_examples.py b/dolphinscheduler-python/pydolphinscheduler/tests/integration/test_submit_examples.py
index 218fa4a55c..393b0cc99a 100644
--- a/dolphinscheduler-python/pydolphinscheduler/tests/integration/test_submit_examples.py
+++ b/dolphinscheduler-python/pydolphinscheduler/tests/integration/test_submit_examples.py
@@ -17,8 +17,8 @@
"""Test whether success submit examples DAG to PythonGatewayService."""
+import subprocess
from pathlib import Path
-from subprocess import Popen
import pytest
@@ -38,7 +38,19 @@ def test_exec_white_list_example(example_path: Path):
"""Test execute examples and submit DAG to PythonGatewayService."""
try:
# Because our task decorator used module ``inspect`` to get the source, and it will
- # raise IOError when call it by built-in function ``exec``, so we change to ``subprocess.Popen``
- Popen(["python", str(example_path)])
- except Exception:
- raise Exception("Run example %s failed.", example_path.stem)
+ # raise IOError when call it by built-in function ``exec``, so we change to ``subprocess.check_call``
+ subprocess.check_call(["python", str(example_path)])
+ except subprocess.CalledProcessError:
+ raise RuntimeError("Run example %s failed.", example_path.stem)
+
+
+def test_exec_multiple_times():
+ """Test whether process definition can be executed more than one times."""
+ tutorial_path = path_example.joinpath("tutorial.py")
+ time = 0
+ while time < 3:
+ try:
+ subprocess.check_call(["python", str(tutorial_path)])
+ except subprocess.CalledProcessError:
+ raise RuntimeError("Run example %s failed.", tutorial_path.stem)
+ time += 1
[dolphinscheduler] 05/06: [Python] Supports creating or editing resources. (#10823)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhongjiajie pushed a commit to branch 3.0.0-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
commit 4da1240b0e1405533708b8b2a1921be6fd2d3bc6
Author: 陈家名 <13...@163.com>
AuthorDate: Tue Jul 12 20:44:59 2022 +0800
[Python] Supports creating or editing resources. (#10823)
(cherry picked from commit 59cd86157ff8d5ab25f0b200d6a30abd95a64c03)
---
.../dolphinscheduler/api/python/PythonGateway.java | 15 +++
.../api/service/ResourcesService.java | 24 ++++
.../api/service/impl/ResourcesServiceImpl.java | 125 ++++++++++++++++++++-
.../api/python/PythonGatewayTest.java | 20 ++++
.../api/service/ResourcesServiceTest.java | 60 +++++++++-
.../apache/dolphinscheduler/common/Constants.java | 5 +
.../src/pydolphinscheduler/core/resource.py | 43 +++++++
.../tests/core/test_process_definition.py | 9 +-
.../tests/core/test_resource_definition.py | 38 +++++++
9 files changed, 331 insertions(+), 8 deletions(-)
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
index 89b396ae98..3dc79a6a5f 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
@@ -549,6 +549,21 @@ public class PythonGateway {
return result;
}
+ /**
+ * create or update resource.
+ * If the folder is not already created, it will be
+ *
+ * @param userName user who create or update resource
+ * @param fullName The fullname of resource.Includes path and suffix.
+ * @param description description of resource
+ * @param resourceContent content of resource
+ * @return id of resource
+ */
+ public Integer createOrUpdateResource(
+ String userName, String fullName, String description, String resourceContent) {
+ return resourceService.createOrUpdateResource(userName, fullName, description, resourceContent);
+ }
+
@PostConstruct
public void init() {
if (pythonGatewayConfiguration.getEnabled()) {
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ResourcesService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ResourcesService.java
index 2433e34a08..113ecb877b 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ResourcesService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/ResourcesService.java
@@ -168,6 +168,30 @@ public interface ResourcesService {
*/
Result<Object> onlineCreateResource(User loginUser, ResourceType type, String fileName, String fileSuffix, String desc, String content,int pid,String currentDirectory);
+ /**
+ * create or update resource.
+ * If the folder is not already created, it will be
+ *
+ * @param loginUser user who create or update resource
+ * @param fileFullName The full name of resource.Includes path and suffix.
+ * @param desc description of resource
+ * @param content content of resource
+ * @return create result code
+ */
+ Result<Object> onlineCreateOrUpdateResourceWithDir(User loginUser, String fileFullName, String desc, String content);
+
+ /**
+ * create or update resource.
+ * If the folder is not already created, it will be
+ *
+ * @param userName user who create or update resource
+ * @param fullName The fullname of resource.Includes path and suffix.
+ * @param description description of resource
+ * @param resourceContent content of resource
+ * @return id of resource
+ */
+ Integer createOrUpdateResource(String userName, String fullName, String description, String resourceContent);
+
/**
* updateProcessInstance resource
*
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java
index 1c5804cd81..cc93f5a77b 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java
@@ -66,16 +66,28 @@ import org.springframework.web.multipart.MultipartFile;
import java.io.IOException;
import java.rmi.ServerException;
import java.text.MessageFormat;
-import java.util.*;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.Date;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.UUID;
import java.util.regex.Matcher;
import java.util.stream.Collectors;
import static org.apache.dolphinscheduler.common.Constants.ALIAS;
import static org.apache.dolphinscheduler.common.Constants.CONTENT;
+import static org.apache.dolphinscheduler.common.Constants.EMPTY_STRING;
import static org.apache.dolphinscheduler.common.Constants.FOLDER_SEPARATOR;
import static org.apache.dolphinscheduler.common.Constants.FORMAT_SS;
import static org.apache.dolphinscheduler.common.Constants.FORMAT_S_S;
import static org.apache.dolphinscheduler.common.Constants.JAR;
+import static org.apache.dolphinscheduler.common.Constants.PERIOD;
/**
* resources service impl
@@ -118,7 +130,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
* @return create directory result
*/
@Override
- @Transactional(rollbackFor = Exception.class)
+ @Transactional
public Result<Object> createDirectory(User loginUser,
String name,
String description,
@@ -189,7 +201,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
* @return create result code
*/
@Override
- @Transactional(rollbackFor = Exception.class)
+ @Transactional
public Result<Object> createResource(User loginUser,
String name,
String desc,
@@ -207,6 +219,12 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
return result;
}
+ // make sure login user has tenant
+ String tenantCode = getTenantCode(loginUser.getId(), result);
+ if (StringUtils.isEmpty(tenantCode)) {
+ return result;
+ }
+
result = verifyFile(name, type, file);
if (!result.getCode().equals(Status.SUCCESS.getCode())) {
return result;
@@ -304,7 +322,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
* @return update result code
*/
@Override
- @Transactional(rollbackFor = Exception.class)
+ @Transactional
public Result<Object> updateResource(User loginUser,
int resourceId,
String name,
@@ -733,7 +751,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
* @throws IOException exception
*/
@Override
- @Transactional(rollbackFor = Exception.class)
+ @Transactional
public Result<Object> delete(User loginUser, int resourceId) throws IOException {
Result<Object> result = checkResourceUploadStartupState();
if (!result.getCode().equals(Status.SUCCESS.getCode())) {
@@ -980,7 +998,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
* @return create result code
*/
@Override
- @Transactional(rollbackFor = Exception.class)
+ @Transactional
public Result<Object> onlineCreateResource(User loginUser, ResourceType type, String fileName, String fileSuffix, String desc, String content, int pid, String currentDir) {
Result<Object> result = checkResourceUploadStartupState();
if (!result.getCode().equals(Status.SUCCESS.getCode())) {
@@ -1035,6 +1053,101 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
return result;
}
+ /**
+ * create or update resource.
+ * If the folder is not already created, it will be
+ *
+ * @param loginUser user who create or update resource
+ * @param fileFullName The full name of resource.Includes path and suffix.
+ * @param desc description of resource
+ * @param content content of resource
+ * @return create result code
+ */
+ @Override
+ @Transactional
+ public Result<Object> onlineCreateOrUpdateResourceWithDir(User loginUser, String fileFullName, String desc, String content) {
+ if (checkResourceExists(fileFullName, ResourceType.FILE.ordinal())) {
+ Resource resource = resourcesMapper.queryResource(fileFullName, ResourceType.FILE.ordinal()).get(0);
+ Result<Object> result = this.updateResourceContent(loginUser, resource.getId(), content);
+ if (result.getCode() == Status.SUCCESS.getCode()) {
+ resource.setDescription(desc);
+ Map<String, Object> resultMap = new HashMap<>();
+ for (Map.Entry<Object, Object> entry : new BeanMap(resource).entrySet()) {
+ if (!Constants.CLASS.equalsIgnoreCase(entry.getKey().toString())) {
+ resultMap.put(entry.getKey().toString(), entry.getValue());
+ }
+ }
+ result.setData(resultMap);
+ }
+ return result;
+ } else {
+ String resourceSuffix = fileFullName.substring(fileFullName.indexOf(PERIOD) + 1);
+ String fileNameWithSuffix = fileFullName.substring(fileFullName.lastIndexOf(FOLDER_SEPARATOR) + 1);
+ String resourceDir = fileFullName.replace(fileNameWithSuffix, EMPTY_STRING);
+ String resourceName = fileNameWithSuffix.replace(PERIOD + resourceSuffix, EMPTY_STRING);
+ String[] dirNames = resourceDir.split(FOLDER_SEPARATOR);
+ int pid = -1;
+ StringBuilder currDirPath = new StringBuilder();
+ for (String dirName : dirNames) {
+ if (StringUtils.isNotEmpty(dirName)) {
+ pid = queryOrCreateDirId(loginUser, pid, currDirPath.toString(), dirName);
+ currDirPath.append(FOLDER_SEPARATOR).append(dirName);
+ }
+ }
+ return this.onlineCreateResource(
+ loginUser, ResourceType.FILE, resourceName, resourceSuffix, desc, content, pid, currDirPath.toString());
+ }
+ }
+
+ @Override
+ @Transactional
+ public Integer createOrUpdateResource(String userName, String fullName, String description, String resourceContent) {
+ User user = userMapper.queryByUserNameAccurately(userName);
+ int suffixLabelIndex = fullName.indexOf(PERIOD);
+ if (suffixLabelIndex == -1) {
+ String msg = String.format("The suffix of file can not be empty : %s", fullName);
+ logger.error(msg);
+ throw new IllegalArgumentException(msg);
+ }
+ if (!fullName.startsWith(FOLDER_SEPARATOR)) {
+ fullName = FOLDER_SEPARATOR + fullName;
+ }
+ Result<Object> createResult = onlineCreateOrUpdateResourceWithDir(
+ user, fullName, description, resourceContent);
+ if (createResult.getCode() == Status.SUCCESS.getCode()) {
+ Map<String, Object> resultMap = (Map<String, Object>) createResult.getData();
+ return (int) resultMap.get("id");
+ }
+ String msg = String.format("Can not create or update resource : %s", fullName);
+ logger.error(msg);
+ throw new IllegalArgumentException(msg);
+ }
+
+ private int queryOrCreateDirId(User user, int pid, String currentDir, String dirName) {
+ String dirFullName = currentDir + FOLDER_SEPARATOR + dirName;
+ if (checkResourceExists(dirFullName, ResourceType.FILE.ordinal())) {
+ List<Resource> resourceList = resourcesMapper.queryResource(dirFullName, ResourceType.FILE.ordinal());
+ return resourceList.get(0).getId();
+ } else {
+ // create dir
+ Result<Object> createDirResult = this.createDirectory(
+ user, dirName, EMPTY_STRING, ResourceType.FILE, pid, currentDir);
+ if (createDirResult.getCode() == Status.SUCCESS.getCode()) {
+ Map<String, Object> resultMap = (Map<String, Object>) createDirResult.getData();
+ return (int) resultMap.get("id");
+ } else {
+ String msg = String.format("Can not create dir %s", dirFullName);
+ logger.error(msg);
+ throw new IllegalArgumentException(msg);
+ }
+ }
+ }
+
+ private void permissionPostHandle(ResourceType resourceType, User loginUser, Integer resourceId) {
+ AuthorizationType authorizationType = resourceType.equals(ResourceType.FILE) ? AuthorizationType.RESOURCE_FILE_ID : AuthorizationType.UDF_FILE;
+ permissionPostHandle(authorizationType, loginUser.getId(), Collections.singletonList(resourceId), logger);
+ }
+
private Result<Object> checkResourceUploadStartupState() {
Result<Object> result = new Result<>();
putMsg(result, Status.SUCCESS);
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java
index 34ff2b6753..022eb344a5 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java
@@ -96,6 +96,26 @@ public class PythonGatewayTest {
Assert.assertEquals((long) result.get("taskDefinitionCode"), taskDefinition.getCode());
}
+ @Test
+ public void testCreateResource() {
+ User user = getTestUser();
+ String resourceDir = "/dir1/dir2/";
+ String resourceName = "test";
+ String resourceSuffix = "py";
+ String desc = "desc";
+ String content = "content";
+ String resourceFullName = resourceDir + resourceName + "." + resourceSuffix;
+
+ int resourceId = 3;
+
+ Mockito.when(resourcesService.createOrUpdateResource(user.getUserName(), resourceFullName, desc, content))
+ .thenReturn(resourceId);
+
+ int id = pythonGateway.createOrUpdateResource(
+ user.getUserName(), resourceFullName, desc, content);
+ Assert.assertEquals(id, resourceId);
+ }
+
@Test
public void testQueryResourcesFileInfo() {
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java
index 939e439f7d..a9642dbfdf 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/ResourcesServiceTest.java
@@ -511,6 +511,64 @@ public class ResourcesServiceTest {
}
+ @Test
+ public void testOnlineCreateResourceWithDir() {
+ User user = getUser();
+ user.setId(1);
+
+ String dir1Path = "/dir1";
+ String dir2Path = "/dir2";
+ String resourceDir = dir1Path + dir2Path;
+ String resourceName = "test";
+ String resourceSuffix = "py";
+ String desc = "desc";
+ String content = "content";
+ String fullName = resourceDir + "/" + resourceName + "." + resourceSuffix;
+
+ Resource dir1 = new Resource();
+ dir1.setFullName(dir1Path);
+ dir1.setId(1);
+ dir1.setUserId(user.getId());
+ Resource dir2 = new Resource();
+ dir2.setFullName(resourceDir);
+ dir2.setUserId(user.getId());
+ Mockito.when(resourcesMapper.queryResource(dir1.getFullName(), ResourceType.FILE.ordinal())).thenReturn(Collections.singletonList(dir1));
+ Mockito.when(resourcesMapper.queryResource(resourceDir, ResourceType.FILE.ordinal())).thenReturn(null);
+ PowerMockito.when(resourcePermissionCheckService.operationPermissionCheck(
+ AuthorizationType.RESOURCE_FILE_ID, 1, ApiFuncIdentificationConstant.FILE_VIEW, serviceLogger)).thenReturn(true);
+ PowerMockito.when(resourcePermissionCheckService.resourcePermissionCheck(
+ AuthorizationType.RESOURCE_FILE_ID, new Object[]{dir1.getId()}, 1, serviceLogger)).thenReturn(true);
+
+ Tenant tenant = getTenant();
+ PowerMockito.when(resourcePermissionCheckService.operationPermissionCheck(
+ AuthorizationType.RESOURCE_FILE_ID, 1, ApiFuncIdentificationConstant.FOLDER_ONLINE_CREATE, serviceLogger)).thenReturn(true);
+ PowerMockito.when(resourcePermissionCheckService.resourcePermissionCheck(
+ AuthorizationType.RESOURCE_FILE_ID, null, 1, serviceLogger)).thenReturn(true);
+ try {
+ PowerMockito.when(storageOperate.mkdir(tenant.getTenantCode(), null)).thenReturn(true);
+ } catch (IOException e) {
+ logger.error("storage error", e);
+ }
+
+ PowerMockito.when(resourcePermissionCheckService.operationPermissionCheck(
+ AuthorizationType.RESOURCE_FILE_ID, 1, ApiFuncIdentificationConstant.FILE_ONLINE_CREATE, serviceLogger)).thenReturn(true);
+ PowerMockito.when(resourcePermissionCheckService.resourcePermissionCheck(
+ AuthorizationType.RESOURCE_FILE_ID, null, 1, serviceLogger)).thenReturn(true);
+ PowerMockito.when(PropertyUtils.getResUploadStartupState()).thenReturn(true);
+ PowerMockito.when(resourcePermissionCheckService.operationPermissionCheck(
+ AuthorizationType.RESOURCE_FILE_ID, 1, ApiFuncIdentificationConstant.FILE_RENAME, serviceLogger)).thenReturn(true);
+ PowerMockito.when(resourcePermissionCheckService.resourcePermissionCheck(
+ AuthorizationType.RESOURCE_FILE_ID, null, 1, serviceLogger)).thenReturn(true);
+ Mockito.when(tenantMapper.queryById(1)).thenReturn(getTenant());
+ Mockito.when(resourcesMapper.selectById(dir1.getId())).thenReturn(dir1);
+ Mockito.when(resourcesMapper.selectById(dir2.getId())).thenReturn(dir2);
+ Mockito.when(FileUtils.getUploadFilename(Mockito.anyString(), Mockito.anyString())).thenReturn("test");
+ PowerMockito.when(FileUtils.writeContent2File(Mockito.anyString(), Mockito.anyString())).thenReturn(true);
+
+ Result<Object> result = resourcesService.onlineCreateOrUpdateResourceWithDir(user, fullName, desc, content);
+ Assert.assertEquals(Status.SUCCESS.getMsg(), result.getMsg());
+ }
+
@Test
public void testUpdateResourceContent() {
PowerMockito.when(PropertyUtils.getResUploadStartupState()).thenReturn(false);
@@ -780,7 +838,7 @@ public class ResourcesServiceTest {
return resource;
}
- private Resource getResource(int resourceId,ResourceType type) {
+ private Resource getResource(int resourceId, ResourceType type) {
Resource resource = new Resource();
resource.setId(resourceId);
diff --git a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
index b4397f007a..6c9c9f3f84 100644
--- a/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
+++ b/dolphinscheduler-common/src/main/java/org/apache/dolphinscheduler/common/Constants.java
@@ -161,6 +161,11 @@ public final class Constants {
*/
public static final String COLON = ":";
+ /**
+ * period .
+ */
+ public static final String PERIOD = ".";
+
/**
* QUESTION ?
*/
diff --git a/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/resource.py b/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/resource.py
new file mode 100644
index 0000000000..bd4ffd4bd7
--- /dev/null
+++ b/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/resource.py
@@ -0,0 +1,43 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+"""Module resource."""
+
+from typing import Optional
+
+from pydolphinscheduler.core.base import Base
+
+
+class Resource(Base):
+ """resource object, will define the resources that you want to create or update.
+
+ :param name: The fullname of resource.Includes path and suffix.
+ :param content: The description of resource.
+ :param description: The description of resource.
+ """
+
+ _DEFINE_ATTR = {"name", "content", "description"}
+
+ def __init__(
+ self,
+ name: str,
+ content: str,
+ description: Optional[str] = None,
+ ):
+ super().__init__(name, description)
+ self.content = content
+ self._resource_code = None
diff --git a/dolphinscheduler-python/pydolphinscheduler/tests/core/test_process_definition.py b/dolphinscheduler-python/pydolphinscheduler/tests/core/test_process_definition.py
index 63580de467..5cb6dabdc0 100644
--- a/dolphinscheduler-python/pydolphinscheduler/tests/core/test_process_definition.py
+++ b/dolphinscheduler-python/pydolphinscheduler/tests/core/test_process_definition.py
@@ -18,7 +18,7 @@
"""Test process definition."""
from datetime import datetime
-from typing import Any
+from typing import Any, List
from unittest.mock import patch
import pytest
@@ -26,6 +26,7 @@ from freezegun import freeze_time
from pydolphinscheduler.core import configuration
from pydolphinscheduler.core.process_definition import ProcessDefinition
+from pydolphinscheduler.core.resource import Resource
from pydolphinscheduler.exceptions import PyDSParamException
from pydolphinscheduler.side import Project, Tenant, User
from pydolphinscheduler.tasks.switch import Branch, Default, Switch, SwitchCondition
@@ -90,6 +91,11 @@ def test_process_definition_default_value(name, value):
("warning_group_id", int, 1),
("timeout", int, 1),
("param", dict, {"key": "value"}),
+ (
+ "resource_list",
+ List,
+ [Resource(name="/dev/test.py", content="hello world", description="desc")],
+ ),
],
)
def test_set_attr(name, cls, expect):
@@ -321,6 +327,7 @@ def test_process_definition_get_define_without_task():
"tasks": {},
"taskDefinitionJson": [{}],
"taskRelationJson": [{}],
+ "resourceList": [],
}
with ProcessDefinition(TEST_PROCESS_DEFINITION_NAME) as pd:
assert pd.get_define() == expect
diff --git a/dolphinscheduler-python/pydolphinscheduler/tests/core/test_resource_definition.py b/dolphinscheduler-python/pydolphinscheduler/tests/core/test_resource_definition.py
new file mode 100644
index 0000000000..ebfb8936f8
--- /dev/null
+++ b/dolphinscheduler-python/pydolphinscheduler/tests/core/test_resource_definition.py
@@ -0,0 +1,38 @@
+# 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.
+
+"""Test resource definition."""
+
+from pydolphinscheduler.core.resource import Resource
+
+
+def test_resource():
+ """Test resource set attributes which get with same type."""
+ name = "/dev/test.py"
+ content = """print("hello world")"""
+ description = "hello world"
+ expect = {
+ "name": name,
+ "content": content,
+ "description": description,
+ }
+ resourceDefinition = Resource(
+ name=name,
+ content=content,
+ description=description,
+ )
+ assert resourceDefinition.get_define() == expect
[dolphinscheduler] 02/06: [python] Integrate test run strategy change (#10825)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhongjiajie pushed a commit to branch 3.0.0-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
commit b2ad6f3f8ec5f6eba7361e3f8c8167f46c45d043
Author: Jiajie Zhong <zh...@hotmail.com>
AuthorDate: Sun Jul 10 11:06:27 2022 +0800
[python] Integrate test run strategy change (#10825)
this patch change python integrate test strategy,
before that, only content change in dir
dolphinscheduler-python will run the test. And it
will fail our python api startter when some code
change in java gateway. So we should start the
integrate test except docs change.
and in the future, we may add python ci to
.asf.yaml to make sure it pass
close: #10488
(cherry picked from commit 598290487c837478cd2afe630cb2f25521e9d5ee)
---
.github/workflows/py-ci.yml | 83 ++++++++++++++++++++++++++-------------------
1 file changed, 48 insertions(+), 35 deletions(-)
diff --git a/.github/workflows/py-ci.yml b/.github/workflows/py-ci.yml
index bec1d74f3f..eddd65b6da 100644
--- a/.github/workflows/py-ci.yml
+++ b/.github/workflows/py-ci.yml
@@ -22,8 +22,6 @@ on:
paths:
- 'dolphinscheduler-python/**'
pull_request:
- paths:
- - 'dolphinscheduler-python/**'
concurrency:
group: py-${{ github.event.pull_request.number || github.ref }}
@@ -38,9 +36,27 @@ env:
DEPENDENCES: pip setuptools wheel tox
jobs:
+ paths-filter:
+ name: Python-Path-Filter
+ runs-on: ubuntu-latest
+ outputs:
+ not-docs: ${{ steps.filter.outputs.not-docs }}
+ py-change: ${{ steps.filter.outputs.py-change }}
+ steps:
+ - uses: actions/checkout@v2
+ - uses: dorny/paths-filter@b2feaf19c27470162a626bd6fa8438ae5b263721
+ id: filter
+ with:
+ filters: |
+ not-docs:
+ - '!(docs/**)'
+ py-change:
+ - 'dolphinscheduler-python/pydolphinscheduler'
lint:
name: Lint
+ if: ${{ (needs.paths-filter.outputs.py-change == 'true') || (github.event_name == 'push') }}
timeout-minutes: 15
+ needs: paths-filter
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v2
@@ -112,14 +128,12 @@ jobs:
- name: Run Tests Build Docs
run: |
python -m tox -vv -e local-ci
- build-image:
- name: Build Image
+ integrate-test:
+ name: Integrate Test
+ if: ${{ (needs.paths-filter.outputs.not-docs == 'true') || (github.event_name == 'push') }}
runs-on: ubuntu-latest
- # Switch to project root directory to run mvnw command
- defaults:
- run:
- working-directory: ./
- timeout-minutes: 20
+ needs: paths-filter
+ timeout-minutes: 30
steps:
- uses: actions/checkout@v2
with:
@@ -132,7 +146,9 @@ jobs:
path: ~/.m2/repository
key: ${{ runner.os }}-maven-${{ hashFiles('**/pom.xml') }}
restore-keys: ${{ runner.os }}-maven-
+ # Switch to project root directory to run mvnw command
- name: Build Image
+ working-directory: ./
run: |
./mvnw -B clean install \
-Dmaven.test.skip \
@@ -140,32 +156,6 @@ jobs:
-Dmaven.checkstyle.skip \
-Pdocker,release -Ddocker.tag=ci \
-pl dolphinscheduler-standalone-server -am
- - name: Export Docker Images
- run: |
- docker save apache/dolphinscheduler-standalone-server:ci -o /tmp/standalone-image.tar \
- && du -sh /tmp/standalone-image.tar
- - uses: actions/upload-artifact@v2
- name: Upload Docker Images
- with:
- name: standalone-image
- path: /tmp/standalone-image.tar
- retention-days: 1
- integrate-test:
- name: Integrate Test
- timeout-minutes: 20
- needs:
- - build-image
- runs-on: ubuntu-latest
- steps:
- - uses: actions/checkout@v2
- - uses: actions/download-artifact@v2
- name: Download Docker Images
- with:
- name: standalone-image
- path: /tmp
- - name: Load Docker Images
- run: |
- docker load -i /tmp/standalone-image.tar
- name: Set up Python 3.7
uses: actions/setup-python@v2
with:
@@ -176,3 +166,26 @@ jobs:
- name: Run Tests Build Docs
run: |
python -m tox -vv -e integrate-test
+ result:
+ name: Python
+ runs-on: ubuntu-latest
+ timeout-minutes: 30
+ needs: [ paths-filter, local-ci, integrate-test ]
+ if: always()
+ steps:
+ - name: Status
+ # We need change CWD to current directory to avoid global default working directory not exists
+ working-directory: ./
+ run: |
+ if [[ ${{ needs.paths-filter.outputs.not-docs }} == 'false' && ${{ github.event_name }} == 'pull_request' ]]; then
+ echo "Only document change, skip both python unit and integrate test!"
+ exit 0
+ fi
+ if [[ ${{ needs.paths-filter.outputs.py-change }} == 'false' && ${{ needs.integrate-test.result }} == 'success' && ${{ github.event_name }} == 'pull_request' ]]; then
+ echo "No python code change, and integrate test pass!"
+ exit 0
+ fi
+ if [[ ${{ needs.integrate-test.result }} != 'success' || ${{ needs.local-ci.result }} != 'success' ]]; then
+ echo "py-ci Failed!"
+ exit -1
+ fi
[dolphinscheduler] 06/06: [fix] Some cherry pick error
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhongjiajie pushed a commit to branch 3.0.0-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
commit 88712b423a23f8c37959978f4263fd3990b841db
Author: Jiajie Zhong <zh...@gmail.com>
AuthorDate: Thu Jul 28 11:12:27 2022 +0800
[fix] Some cherry pick error
---
.../dolphinscheduler/api/python/PythonGateway.java | 2 +-
.../api/service/impl/QueueServiceImpl.java | 11 ------
.../api/service/impl/ResourcesServiceImpl.java | 9 +----
.../api/service/impl/TenantServiceImpl.java | 43 +++++++---------------
.../api/python/PythonGatewayTest.java | 2 +-
.../pydolphinscheduler/core/process_definition.py | 1 -
6 files changed, 18 insertions(+), 50 deletions(-)
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
index 3dc79a6a5f..67cbeb81a8 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
@@ -537,7 +537,7 @@ public class PythonGateway {
public Map<String, Object> queryResourcesFileInfo(String userName, String fullName) {
Map<String, Object> result = new HashMap<>();
User user = usersService.queryUser(userName);
- Result<Object> resourceResponse = resourceService.queryResource(user, fullName, null, ResourceType.FILE);
+ Result<Object> resourceResponse = resourceService.queryResource(fullName, null, ResourceType.FILE);
if (resourceResponse.getCode() != Status.SUCCESS.getCode()) {
String msg = String.format("Can not find valid resource by name %s", fullName);
logger.error(msg);
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
index bce236250e..435f3875e2 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
@@ -17,17 +17,12 @@
package org.apache.dolphinscheduler.api.service.impl;
-import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_CREATE;
-import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_UPDATE;
-
import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.QueueService;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
-import org.apache.dolphinscheduler.common.enums.AuthorizationType;
-import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.dao.entity.Queue;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.QueueMapper;
@@ -35,20 +30,15 @@ import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.commons.lang3.StringUtils;
-import java.util.ArrayList;
-import java.util.Collections;
import java.util.HashMap;
-import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Objects;
-import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
-import org.springframework.transaction.annotation.Transactional;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@@ -181,7 +171,6 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
result.put(Constants.DATA_LIST, queueObj);
putMsg(result, Status.SUCCESS);
- permissionPostHandle(AuthorizationType.QUEUE, loginUser.getId(), Collections.singletonList(queueObj.getId()), logger);
return result;
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java
index cc93f5a77b..312bb014d4 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java
@@ -341,7 +341,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
return result;
}
- if (!PropertyUtils.getResUploadStartupState()){
+ if (!PropertyUtils.getResUploadStartupState()) {
putMsg(result, Status.STORAGE_NOT_STARTUP);
return result;
}
@@ -1068,7 +1068,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
public Result<Object> onlineCreateOrUpdateResourceWithDir(User loginUser, String fileFullName, String desc, String content) {
if (checkResourceExists(fileFullName, ResourceType.FILE.ordinal())) {
Resource resource = resourcesMapper.queryResource(fileFullName, ResourceType.FILE.ordinal()).get(0);
- Result<Object> result = this.updateResourceContent(loginUser, resource.getId(), content);
+ Result<Object> result = this.updateResourceContent(resource.getId(), content);
if (result.getCode() == Status.SUCCESS.getCode()) {
resource.setDescription(desc);
Map<String, Object> resultMap = new HashMap<>();
@@ -1143,11 +1143,6 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
}
}
- private void permissionPostHandle(ResourceType resourceType, User loginUser, Integer resourceId) {
- AuthorizationType authorizationType = resourceType.equals(ResourceType.FILE) ? AuthorizationType.RESOURCE_FILE_ID : AuthorizationType.UDF_FILE;
- permissionPostHandle(authorizationType, loginUser.getId(), Collections.singletonList(resourceId), logger);
- }
-
private Result<Object> checkResourceUploadStartupState() {
Result<Object> result = new Result<>();
putMsg(result, Status.SUCCESS);
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
index f07966af76..559df56dca 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
@@ -29,7 +29,6 @@ import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.RegexUtils;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
-import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.storage.StorageOperate;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
@@ -47,16 +46,11 @@ import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
-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 java.util.Objects;
-import java.util.Set;
-import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.*;
import static org.apache.dolphinscheduler.common.Constants.TENANT_FULL_NAME_MAX_LENGTH;
/**
@@ -143,8 +137,8 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
String desc) throws Exception {
Map<String, Object> result = new HashMap<>();
result.put(Constants.STATUS, false);
- if (!canOperatorPermissions(loginUser,null, AuthorizationType.TENANT, TENANT_CREATE)) {
- throw new ServiceException(Status.USER_NO_OPERATION_PERM);
+ if (isNotAdmin(loginUser, result)) {
+ return result;
}
Tenant tenant = new Tenant(tenantCode, desc, queueId);
@@ -155,7 +149,6 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
if (PropertyUtils.getResUploadStartupState()) {
storageOperate.createTenantDirIfNotExists(tenantCode);
}
- permissionPostHandle(AuthorizationType.TENANT, loginUser.getId(), Collections.singletonList(tenant.getId()), logger);
result.put(Constants.DATA_LIST, tenant);
putMsg(result, Status.SUCCESS);
return result;
@@ -174,18 +167,15 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
public Result<Object> queryTenantList(User loginUser, String searchVal, Integer pageNo, Integer pageSize) {
Result<Object> result = new Result<>();
- PageInfo<Tenant> pageInfo = new PageInfo<>(pageNo, pageSize);
- Set<Integer> ids = resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.TENANT, loginUser.getId(), logger);
- if (ids.isEmpty()) {
- result.setData(pageInfo);
- putMsg(result, Status.SUCCESS);
- return result;
+ if (!isAdmin(loginUser)) {
+ putMsg(result, Status.USER_NO_OPERATION_PERM);
}
Page<Tenant> page = new Page<>(pageNo, pageSize);
- IPage<Tenant> tenantPage = tenantMapper.queryTenantPaging(page, new ArrayList<>(ids), searchVal);
+ IPage<Tenant> tenantIPage = tenantMapper.queryTenantPaging(page, searchVal);
+ PageInfo<Tenant> pageInfo = new PageInfo<>(pageNo, pageSize);
+ pageInfo.setTotal((int) tenantIPage.getTotal());
+ pageInfo.setTotalList(tenantIPage.getRecords());
- pageInfo.setTotal((int) tenantPage.getTotal());
- pageInfo.setTotalList(tenantPage.getRecords());
result.setData(pageInfo);
putMsg(result, Status.SUCCESS);
return result;
@@ -207,9 +197,10 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
String desc) throws Exception {
Map<String, Object> result = new HashMap<>();
+ result.put(Constants.STATUS, false);
- if (!canOperatorPermissions(loginUser,null, AuthorizationType.TENANT,TENANT_UPDATE)) {
- throw new ServiceException(Status.USER_NO_OPERATION_PERM);
+ if (isNotAdmin(loginUser, result)) {
+ return result;
}
Tenant updateTenant = new Tenant(id, tenantCode, desc, queueId);
@@ -242,8 +233,8 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
public Map<String, Object> deleteTenantById(User loginUser, int id) throws Exception {
Map<String, Object> result = new HashMap<>();
- if (!canOperatorPermissions(loginUser,null, AuthorizationType.TENANT,TENANT_DELETE)) {
- throw new ServiceException(Status.USER_NO_OPERATION_PERM);
+ if (isNotAdmin(loginUser, result)) {
+ return result;
}
Tenant tenant = tenantMapper.queryById(id);
@@ -293,13 +284,7 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
public Map<String, Object> queryTenantList(User loginUser) {
Map<String, Object> result = new HashMap<>();
- Set<Integer> ids = resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.TENANT, loginUser.getId(), logger);
- if (ids.isEmpty()) {
- result.put(Constants.DATA_LIST, Collections.emptyList());
- putMsg(result, Status.SUCCESS);
- return result;
- }
- List<Tenant> resourceList = tenantMapper.selectBatchIds(ids);
+ List<Tenant> resourceList = tenantMapper.selectList(null);
result.put(Constants.DATA_LIST, resourceList);
putMsg(result, Status.SUCCESS);
return result;
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java
index 022eb344a5..a2c63a49f3 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/python/PythonGatewayTest.java
@@ -126,7 +126,7 @@ public class PythonGatewayTest {
mockResult.setCode(Status.SUCCESS.getCode());
Resource resource = getTestResource();
mockResult.setData(resource);
- Mockito.when(resourcesService.queryResource(user, resource.getFullName(), null, ResourceType.FILE)).thenReturn(mockResult);
+ Mockito.when(resourcesService.queryResource(resource.getFullName(), null, ResourceType.FILE)).thenReturn(mockResult);
Map<String, Object> result = pythonGateway.queryResourcesFileInfo(user.getUserName(), resource.getFullName());
Assert.assertEquals((int) result.get("id"), resource.getId());
diff --git a/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/process_definition.py b/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/process_definition.py
index dd18dafa0f..63e080899d 100644
--- a/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/process_definition.py
+++ b/dolphinscheduler-python/pydolphinscheduler/src/pydolphinscheduler/core/process_definition.py
@@ -411,7 +411,6 @@ class ProcessDefinition(Base):
json.dumps(self.task_relation_json),
json.dumps(self.task_definition_json),
None,
- None,
)
if len(self.resource_list) > 0:
for res in self.resource_list:
[dolphinscheduler] 01/06: [python] Fix submit and run error (#10792)
Posted by zh...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zhongjiajie pushed a commit to branch 3.0.0-prepare
in repository https://gitbox.apache.org/repos/asf/dolphinscheduler.git
commit d3f28c84112169f12d14863450cc1c4df0c25267
Author: Jiajie Zhong <zh...@gmail.com>
AuthorDate: Fri Jul 8 10:54:41 2022 +0800
[python] Fix submit and run error (#10792)
because we add permission, so it failed out
createQueue method. this patch fix it and
do some refactor of our tenant and queue
validator code
(cherry picked from commit ae6aa53f963d2ea02ab435db1e5548da23e3f882)
---
.../apache/dolphinscheduler/api/enums/Status.java | 2 +-
.../api/exceptions/ApiExceptionHandler.java | 6 +
.../api/exceptions/ServiceException.java | 7 +
.../dolphinscheduler/api/python/PythonGateway.java | 37 +---
.../dolphinscheduler/api/service/QueueService.java | 10 +-
.../api/service/TenantService.java | 18 +-
.../api/service/impl/QueueServiceImpl.java | 187 ++++++++-----------
.../api/service/impl/TenantServiceImpl.java | 205 ++++++++++++--------
.../api/controller/QueueControllerTest.java | 60 ++++--
.../api/service/QueueServiceTest.java | 144 +++++++-------
.../api/service/TenantServiceTest.java | 207 ++++++++++++---------
.../dolphinscheduler/dao/mapper/QueueMapper.java | 7 +-
.../dolphinscheduler/dao/mapper/QueueMapper.xml | 12 +-
13 files changed, 487 insertions(+), 415 deletions(-)
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
index 7017fa6d95..364f8f9d8e 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/enums/Status.java
@@ -194,7 +194,7 @@ public enum Status {
QUERY_WORKFLOW_LINEAGE_ERROR(10161, "query workflow lineage error", "查询血缘失败"),
QUERY_AUTHORIZED_AND_USER_CREATED_PROJECT_ERROR(10162, "query authorized and user created project error error", "查询授权的和用户创建的项目错误"),
DELETE_PROCESS_DEFINITION_BY_CODE_FAIL(10163, "delete process definition by code fail, for there are {0} process instances in executing using it", "删除工作流定义失败，有[{0}]个运行中的工作流实例正在使用"),
- CHECK_OS_TENANT_CODE_ERROR(10164, "Please enter the English os tenant code", "请输入英文操作系统租户"),
+ CHECK_OS_TENANT_CODE_ERROR(10164, "Tenant code invalid, should follow linux's users naming conventions", "非法的租户名，需要遵守 Linux 用户命名规范"),
FORCE_TASK_SUCCESS_ERROR(10165, "force task success error", "强制成功任务实例错误"),
TASK_INSTANCE_STATE_OPERATION_ERROR(10166, "the status of task instance {0} is {1},Cannot perform force success operation", "任务实例[{0}]的状态是[{1}]，无法执行强制成功操作"),
DATASOURCE_TYPE_NOT_EXIST(10167, "data source type not exist", "数据源类型不存在"),
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ApiExceptionHandler.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ApiExceptionHandler.java
index 84d1abc337..e1d5a47138 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ApiExceptionHandler.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ApiExceptionHandler.java
@@ -36,6 +36,12 @@ public class ApiExceptionHandler {
private static final Logger logger = LoggerFactory.getLogger(ApiExceptionHandler.class);
+ @ExceptionHandler(ServiceException.class)
+ public Result exceptionHandler(ServiceException e, HandlerMethod hm) {
+ logger.error("ServiceException: ", e);
+ return new Result(e.getCode(), e.getMessage());
+ }
+
@ExceptionHandler(Exception.class)
public Result exceptionHandler(Exception e, HandlerMethod hm) {
ApiException ce = hm.getMethodAnnotation(ApiException.class);
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ServiceException.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ServiceException.java
index 932b78f632..369fbbec3e 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ServiceException.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/exceptions/ServiceException.java
@@ -18,6 +18,8 @@ package org.apache.dolphinscheduler.api.exceptions;
import org.apache.dolphinscheduler.api.enums.Status;
+import java.text.MessageFormat;
+
/**
* service exception
@@ -37,6 +39,11 @@ public class ServiceException extends RuntimeException {
this.code = status.getCode();
}
+ public ServiceException(Status status, Object... formatter) {
+ super(MessageFormat.format(status.getMsg(), formatter));
+ this.code = status.getCode();
+ }
+
public ServiceException(Integer code,String message) {
super(message);
this.code = code;
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
index 7f474fa689..71db35cb2a 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/python/PythonGateway.java
@@ -22,7 +22,6 @@ import org.apache.dolphinscheduler.api.enums.Status;
import org.apache.dolphinscheduler.api.service.ExecutorService;
import org.apache.dolphinscheduler.api.service.ProcessDefinitionService;
import org.apache.dolphinscheduler.api.service.ProjectService;
-import org.apache.dolphinscheduler.api.service.QueueService;
import org.apache.dolphinscheduler.api.service.ResourcesService;
import org.apache.dolphinscheduler.api.service.SchedulerService;
import org.apache.dolphinscheduler.api.service.TaskDefinitionService;
@@ -114,9 +113,6 @@ public class PythonGateway {
@Autowired
private UsersService usersService;
- @Autowired
- private QueueService queueService;
-
@Autowired
private ResourcesService resourceService;
@@ -395,37 +391,8 @@ public class PythonGateway {
}
}
- public Map<String, Object> createQueue(String name, String queueName) {
- Result<Object> verifyQueueExists = queueService.verifyQueue(name, queueName);
- if (verifyQueueExists.getCode() == 0) {
- return queueService.createQueue(dummyAdminUser, name, queueName);
- } else {
- Map<String, Object> result = new HashMap<>();
- // TODO function putMsg do not work here
- result.put(Constants.STATUS, Status.SUCCESS);
- result.put(Constants.MSG, Status.SUCCESS.getMsg());
- return result;
- }
- }
-
- public Map<String, Object> createTenant(String tenantCode, String desc, String queueName) throws Exception {
- if (tenantService.checkTenantExists(tenantCode)) {
- Map<String, Object> result = new HashMap<>();
- // TODO function putMsg do not work here
- result.put(Constants.STATUS, Status.SUCCESS);
- result.put(Constants.MSG, Status.SUCCESS.getMsg());
- return result;
- } else {
- Result<Object> verifyQueueExists = queueService.verifyQueue(queueName, queueName);
- if (verifyQueueExists.getCode() == 0) {
- // TODO why create do not return id?
- queueService.createQueue(dummyAdminUser, queueName, queueName);
- }
- Map<String, Object> result = queueService.queryQueueName(queueName);
- List<Queue> queueList = (List<Queue>) result.get(Constants.DATA_LIST);
- Queue queue = queueList.get(0);
- return tenantService.createTenant(dummyAdminUser, tenantCode, queue.getId(), desc);
- }
+ public Tenant createTenant(String tenantCode, String desc, String queueName) {
+ return tenantService.createTenantIfNotExists(tenantCode, desc, queueName, queueName);
}
public void createUser(String userName,
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/QueueService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/QueueService.java
index f978b96b6b..a29fc465bd 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/QueueService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/QueueService.java
@@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.api.service;
import org.apache.dolphinscheduler.api.utils.Result;
+import org.apache.dolphinscheduler.dao.entity.Queue;
import org.apache.dolphinscheduler.dao.entity.User;
import java.util.Map;
@@ -77,11 +78,14 @@ public interface QueueService {
Result<Object> verifyQueue(String queue, String queueName);
/**
- * query queue by queueName
+ * Make sure queue with given name exists, and create the queue if not exists
*
+ * ONLY for python gateway server, and should not use this in web ui function
+ *
+ * @param queue queue value
* @param queueName queue name
- * @return queue object for provide queue name
+ * @return Queue object
*/
- Map<String, Object> queryQueueName(String queueName);
+ Queue createQueueIfNotExists(String queue, String queueName);
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TenantService.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TenantService.java
index 47a4082a30..0d795a10c4 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TenantService.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/TenantService.java
@@ -18,6 +18,7 @@
package org.apache.dolphinscheduler.api.service;
import org.apache.dolphinscheduler.api.utils.Result;
+import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.User;
import java.util.Map;
@@ -94,18 +95,23 @@ public interface TenantService {
Result verifyTenantCode(String tenantCode);
/**
- * check if provide tenant code object exists
+ * query tenant by tenant code
*
* @param tenantCode tenant code
- * @return true if tenant code exists, false if not
+ * @return tenant list
*/
- boolean checkTenantExists(String tenantCode);
+ Map<String, Object> queryByTenantCode(String tenantCode);
/**
- * query tenant by tenant code
+ * Make sure tenant with given name exists, and create the tenant if not exists
+ *
+ * ONLY for python gateway server, and should not use this in web ui function
*
* @param tenantCode tenant code
- * @return tenant list
+ * @param desc The description of tenant object
+ * @param queue The value of queue which current tenant belong
+ * @param queueName The name of queue which current tenant belong
+ * @return Tenant object
*/
- Map<String, Object> queryByTenantCode(String tenantCode);
+ Tenant createTenantIfNotExists(String tenantCode, String desc, String queue, String queueName);
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
index 2da89df000..317bcd2cee 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/QueueServiceImpl.java
@@ -17,30 +17,40 @@
package org.apache.dolphinscheduler.api.service.impl;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import org.apache.commons.lang3.StringUtils;
import org.apache.dolphinscheduler.api.enums.Status;
+import org.apache.dolphinscheduler.api.exceptions.ServiceException;
import org.apache.dolphinscheduler.api.service.QueueService;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.AuthorizationType;
+import org.apache.dolphinscheduler.common.enums.UserType;
import org.apache.dolphinscheduler.dao.entity.Queue;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.QueueMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
-import org.apache.commons.lang.StringUtils;
-
+import java.util.ArrayList;
+import java.util.Collections;
import java.util.Date;
import java.util.HashMap;
+import java.util.HashSet;
import java.util.List;
import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
+import org.springframework.transaction.annotation.Transactional;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_CREATE;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_UPDATE;
/**
* queue service impl
@@ -56,6 +66,44 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
@Autowired
private UserMapper userMapper;
+ /**
+ * Valid both queue and queueName when we want to create or update queue object
+ *
+ * @param queue queue value
+ * @param queueName queue name
+ */
+ private void queueValid(String queue, String queueName) throws ServiceException {
+ if (StringUtils.isEmpty(queue)) {
+ throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
+ } else if (StringUtils.isEmpty(queueName)) {
+ throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
+ } else if (checkQueueExist(queue)) {
+ throw new ServiceException(Status.QUEUE_VALUE_EXIST, queue);
+ } else if (checkQueueNameExist(queueName)) {
+ throw new ServiceException(Status.QUEUE_NAME_EXIST, queueName);
+ }
+ }
+
+ /**
+ * Insert one single new Queue record to database
+ *
+ * @param queue queue value
+ * @param queueName queue name
+ * @return Queue
+ */
+ private Queue createObjToDB(String queue, String queueName) {
+ Queue queueObj = new Queue();
+ Date now = new Date();
+
+ queueObj.setQueue(queue);
+ queueObj.setQueueName(queueName);
+ queueObj.setCreateTime(now);
+ queueObj.setUpdateTime(now);
+ // save
+ queueMapper.insert(queueObj);
+ return queueObj;
+ }
+
/**
* query queue list
*
@@ -121,39 +169,11 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
if (isNotAdmin(loginUser, result)) {
return result;
}
+ queueValid(queue, queueName);
- if (StringUtils.isEmpty(queue)) {
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
- return result;
- }
-
- if (StringUtils.isEmpty(queueName)) {
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
- return result;
- }
-
- if (checkQueueNameExist(queueName)) {
- putMsg(result, Status.QUEUE_NAME_EXIST, queueName);
- return result;
- }
-
- if (checkQueueExist(queue)) {
- putMsg(result, Status.QUEUE_VALUE_EXIST, queue);
- return result;
- }
-
- Queue queueObj = new Queue();
- Date now = new Date();
-
- queueObj.setQueue(queue);
- queueObj.setQueueName(queueName);
- queueObj.setCreateTime(now);
- queueObj.setUpdateTime(now);
-
- queueMapper.insert(queueObj);
- result.put(Constants.DATA_LIST, queueObj);
+ Queue newQueue = createObjToDB(queue, queueName);
+ result.put(Constants.DATA_LIST, newQueue);
putMsg(result, Status.SUCCESS);
-
return result;
}
@@ -173,39 +193,16 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
return result;
}
- if (StringUtils.isEmpty(queue)) {
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
- return result;
- }
-
- if (StringUtils.isEmpty(queueName)) {
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
- return result;
- }
+ queueValid(queue, queueName);
Queue queueObj = queueMapper.selectById(id);
- if (queueObj == null) {
- putMsg(result, Status.QUEUE_NOT_EXIST, id);
- return result;
+ if (Objects.isNull(queueObj)) {
+ throw new ServiceException(Status.QUEUE_NOT_EXIST, queue);
}
// whether queue value or queueName is changed
if (queue.equals(queueObj.getQueue()) && queueName.equals(queueObj.getQueueName())) {
- putMsg(result, Status.NEED_NOT_UPDATE_QUEUE);
- return result;
- }
-
- // check queue name is exist
- if (!queueName.equals(queueObj.getQueueName())
- && checkQueueNameExist(queueName)) {
- putMsg(result, Status.QUEUE_NAME_EXIST, queueName);
- return result;
- }
-
- // check queue value is exist
- if (!queue.equals(queueObj.getQueue()) && checkQueueExist(queue)) {
- putMsg(result, Status.QUEUE_VALUE_EXIST, queue);
- return result;
+ throw new ServiceException(Status.NEED_NOT_UPDATE_QUEUE);
}
// check old queue using by any user
@@ -238,53 +235,8 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
@Override
public Result<Object> verifyQueue(String queue, String queueName) {
Result<Object> result = new Result<>();
+ queueValid(queue, queueName);
- if (StringUtils.isEmpty(queue)) {
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE);
- return result;
- }
-
- if (StringUtils.isEmpty(queueName)) {
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
- return result;
- }
-
- if (checkQueueNameExist(queueName)) {
- putMsg(result, Status.QUEUE_NAME_EXIST, queueName);
- return result;
- }
-
- if (checkQueueExist(queue)) {
- putMsg(result, Status.QUEUE_VALUE_EXIST, queue);
- return result;
- }
-
- putMsg(result, Status.SUCCESS);
- return result;
- }
-
- /**
- * query queue by queueName
- *
- * @param queueName queue name
- * @return queue object for provide queue name
- */
- @Override
- public Map<String, Object> queryQueueName(String queueName) {
- Map<String, Object> result = new HashMap<>();
-
- if (StringUtils.isEmpty(queueName)) {
- putMsg(result, Status.REQUEST_PARAMS_NOT_VALID_ERROR, Constants.QUEUE_NAME);
- return result;
- }
-
- if (!checkQueueNameExist(queueName)) {
- putMsg(result, Status.QUEUE_NOT_EXIST, queueName);
- return result;
- }
-
- List<Queue> queueList = queueMapper.queryQueueName(queueName);
- result.put(Constants.DATA_LIST, queueList);
putMsg(result, Status.SUCCESS);
return result;
}
@@ -324,4 +276,23 @@ public class QueueServiceImpl extends BaseServiceImpl implements QueueService {
return !oldQueue.equals(newQueue) && userMapper.existUser(oldQueue) == Boolean.TRUE;
}
+ /**
+ * Make sure queue with given name exists, and create the queue if not exists
+ *
+ * ONLY for python gateway server, and should not use this in web ui function
+ *
+ * @param queue queue value
+ * @param queueName queue name
+ * @return Queue object
+ */
+ @Override
+ public Queue createQueueIfNotExists(String queue, String queueName) {
+ queueValid(queue, queueName);
+ Queue existsQueue = queueMapper.queryQueueName(queue, queueName);
+ if (Objects.isNull(existsQueue)) {
+ return createObjToDB(queue, queueName);
+ }
+ return existsQueue;
+ }
+
}
diff --git a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
index 6bb2870452..c4cca300fb 100644
--- a/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
+++ b/dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/TenantServiceImpl.java
@@ -20,31 +20,43 @@ package org.apache.dolphinscheduler.api.service.impl;
import com.baomidou.mybatisplus.core.metadata.IPage;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import org.apache.commons.collections.CollectionUtils;
-import org.apache.commons.lang.StringUtils;
+import org.apache.commons.lang3.StringUtils;
import org.apache.dolphinscheduler.api.enums.Status;
+import org.apache.dolphinscheduler.api.exceptions.ServiceException;
+import org.apache.dolphinscheduler.api.service.QueueService;
import org.apache.dolphinscheduler.api.service.TenantService;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.RegexUtils;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.storage.StorageOperate;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
+import org.apache.dolphinscheduler.dao.entity.Queue;
import org.apache.dolphinscheduler.dao.entity.Tenant;
import org.apache.dolphinscheduler.dao.entity.User;
import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
+
+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 java.util.Objects;
+import java.util.Set;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.*;
import static org.apache.dolphinscheduler.common.Constants.TENANT_FULL_NAME_MAX_LENGTH;
/**
@@ -53,6 +65,8 @@ import static org.apache.dolphinscheduler.common.Constants.TENANT_FULL_NAME_MAX_
@Service
public class TenantServiceImpl extends BaseServiceImpl implements TenantService {
+ private static final Logger logger = LoggerFactory.getLogger(TenantServiceImpl.class);
+
@Autowired
private TenantMapper tenantMapper;
@@ -65,9 +79,53 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
@Autowired
private UserMapper userMapper;
+ @Autowired
+ private QueueService queueService;
+
@Autowired(required = false)
private StorageOperate storageOperate;
+ /**
+ * Valid tenantCode when we want to create or update tenant object
+ *
+ * @param tenantCode Tenant code of tenant object
+ * @return Optional of Status map
+ */
+ private void tenantCodeValid(String tenantCode) throws ServiceException {
+ Map<String, Object> result = new HashMap<>();
+ if (StringUtils.isEmpty(tenantCode)) {
+ throw new ServiceException(Status.REQUEST_PARAMS_NOT_VALID_ERROR, tenantCode);
+ } else if (StringUtils.length(tenantCode) > TENANT_FULL_NAME_MAX_LENGTH) {
+ throw new ServiceException(Status.TENANT_FULL_NAME_TOO_LONG_ERROR);
+ } else if (!RegexUtils.isValidLinuxUserName(tenantCode)) {
+ throw new ServiceException(Status.CHECK_OS_TENANT_CODE_ERROR);
+ } else if (checkTenantExists(tenantCode)) {
+ throw new ServiceException(Status.OS_TENANT_CODE_EXIST, tenantCode);
+ }
+ }
+
+ /**
+ * Insert one single new Tenant record to database
+ *
+ * @param tenantCode new Tenant object tenant code
+ * @param desc new Tenant object description
+ * @param queueId The Queue id of new Tenant object
+ * @return Tenant
+ */
+ private Tenant createObjToDB(String tenantCode, String desc, int queueId) {
+ Tenant tenant = new Tenant();
+ Date now = new Date();
+
+ tenant.setTenantCode(tenantCode);
+ tenant.setQueueId(queueId);
+ tenant.setDescription(desc);
+ tenant.setCreateTime(now);
+ tenant.setUpdateTime(now);
+ // save
+ tenantMapper.insert(tenant);
+ return tenant;
+ }
+
/**
* create tenant
*
@@ -79,51 +137,27 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
* @throws Exception exception
*/
@Override
- @Transactional(rollbackFor = Exception.class)
+ @Transactional
public Map<String, Object> createTenant(User loginUser,
String tenantCode,
int queueId,
String desc) throws Exception {
Map<String, Object> result = new HashMap<>();
result.put(Constants.STATUS, false);
- if (isNotAdmin(loginUser, result)) {
- return result;
- }
-
- if(StringUtils.length(tenantCode) > TENANT_FULL_NAME_MAX_LENGTH){
- putMsg(result, Status.TENANT_FULL_NAME_TOO_LONG_ERROR);
- return result;
- }
-
- if (!RegexUtils.isValidLinuxUserName(tenantCode)) {
- putMsg(result, Status.CHECK_OS_TENANT_CODE_ERROR);
- return result;
+ if (!canOperatorPermissions(loginUser,null, AuthorizationType.TENANT, TENANT_CREATE)) {
+ throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
+ tenantCodeValid(tenantCode);
- if (checkTenantExists(tenantCode)) {
- putMsg(result, Status.OS_TENANT_CODE_EXIST, tenantCode);
- return result;
- }
-
- Tenant tenant = new Tenant();
- Date now = new Date();
- tenant.setTenantCode(tenantCode);
- tenant.setQueueId(queueId);
- tenant.setDescription(desc);
- tenant.setCreateTime(now);
- tenant.setUpdateTime(now);
- // save
- tenantMapper.insert(tenant);
-
+ Tenant newTenant = createObjToDB(tenantCode, desc, queueId);
// if storage startup
if (PropertyUtils.getResUploadStartupState()) {
storageOperate.createTenantDirIfNotExists(tenantCode);
}
-
- result.put(Constants.DATA_LIST, tenant);
+ permissionPostHandle(AuthorizationType.TENANT, loginUser.getId(), Collections.singletonList(newTenant.getId()), logger);
+ result.put(Constants.DATA_LIST, newTenant);
putMsg(result, Status.SUCCESS);
-
return result;
}
@@ -140,16 +174,18 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
public Result<Object> queryTenantList(User loginUser, String searchVal, Integer pageNo, Integer pageSize) {
Result<Object> result = new Result<>();
- if (!isAdmin(loginUser)) {
- putMsg(result, Status.USER_NO_OPERATION_PERM);
+ PageInfo<Tenant> pageInfo = new PageInfo<>(pageNo, pageSize);
+ Set<Integer> ids = resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.TENANT, loginUser.getId(), logger);
+ if (ids.isEmpty()) {
+ result.setData(pageInfo);
+ putMsg(result, Status.SUCCESS);
return result;
}
-
Page<Tenant> page = new Page<>(pageNo, pageSize);
- IPage<Tenant> tenantIPage = tenantMapper.queryTenantPaging(page, searchVal);
- PageInfo<Tenant> pageInfo = new PageInfo<>(pageNo, pageSize);
- pageInfo.setTotal((int) tenantIPage.getTotal());
- pageInfo.setTotalList(tenantIPage.getRecords());
+ IPage<Tenant> tenantPage = tenantMapper.queryTenantPaging(page, new ArrayList<>(ids), searchVal);
+
+ pageInfo.setTotal((int) tenantPage.getTotal());
+ pageInfo.setTotalList(tenantPage.getRecords());
result.setData(pageInfo);
putMsg(result, Status.SUCCESS);
return result;
@@ -171,41 +207,29 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
String desc) throws Exception {
Map<String, Object> result = new HashMap<>();
- result.put(Constants.STATUS, false);
- if (isNotAdmin(loginUser, result)) {
- return result;
+ if (!canOperatorPermissions(loginUser,null, AuthorizationType.TENANT,TENANT_UPDATE)) {
+ throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
Tenant tenant = tenantMapper.queryById(id);
- if (tenant == null) {
- putMsg(result, Status.TENANT_NOT_EXIST);
- return result;
+ if (Objects.isNull(tenant)) {
+ throw new ServiceException(Status.TENANT_NOT_EXIST);
}
+ tenantCodeValid(tenantCode);
+
// updateProcessInstance tenant
/**
* if the tenant code is modified, the original resource needs to be copied to the new tenant.
*/
- if (!tenant.getTenantCode().equals(tenantCode)) {
- if (checkTenantExists(tenantCode)) {
- // if hdfs startup
- if (PropertyUtils.getResUploadStartupState()) {
- storageOperate.createTenantDirIfNotExists(tenantCode);
- }
- } else {
- putMsg(result, Status.OS_TENANT_CODE_HAS_ALREADY_EXISTS);
- return result;
- }
+ if (!tenant.getTenantCode().equals(tenantCode) && PropertyUtils.getResUploadStartupState()) {
+ storageOperate.createTenantDirIfNotExists(tenantCode);
}
Date now = new Date();
- if (!StringUtils.isEmpty(tenantCode)) {
- tenant.setTenantCode(tenantCode);
- }
-
if (queueId != 0) {
tenant.setQueueId(queueId);
}
@@ -213,8 +237,7 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
tenant.setUpdateTime(now);
tenantMapper.updateById(tenant);
- result.put(Constants.STATUS, Status.SUCCESS);
- result.put(Constants.MSG, Status.SUCCESS.getMsg());
+ putMsg(result, Status.SUCCESS);
return result;
}
@@ -227,37 +250,33 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
* @throws Exception exception
*/
@Override
- @Transactional(rollbackFor = Exception.class)
+ @Transactional
public Map<String, Object> deleteTenantById(User loginUser, int id) throws Exception {
Map<String, Object> result = new HashMap<>();
- if (isNotAdmin(loginUser, result)) {
- return result;
+ if (!canOperatorPermissions(loginUser,null, AuthorizationType.TENANT,TENANT_DELETE)) {
+ throw new ServiceException(Status.USER_NO_OPERATION_PERM);
}
Tenant tenant = tenantMapper.queryById(id);
- if (tenant == null) {
- putMsg(result, Status.TENANT_NOT_EXIST);
- return result;
+ if (Objects.isNull(tenant)) {
+ throw new ServiceException(Status.TENANT_NOT_EXIST);
}
List<ProcessInstance> processInstances = getProcessInstancesByTenant(tenant);
if (CollectionUtils.isNotEmpty(processInstances)) {
- putMsg(result, Status.DELETE_TENANT_BY_ID_FAIL, processInstances.size());
- return result;
+ throw new ServiceException(Status.DELETE_TENANT_BY_ID_FAIL, processInstances.size());
}
List<ProcessDefinition> processDefinitions =
processDefinitionMapper.queryDefinitionListByTenant(tenant.getId());
if (CollectionUtils.isNotEmpty(processDefinitions)) {
- putMsg(result, Status.DELETE_TENANT_BY_ID_FAIL_DEFINES, processDefinitions.size());
- return result;
+ throw new ServiceException(Status.DELETE_TENANT_BY_ID_FAIL_DEFINES, processDefinitions.size());
}
List<User> userList = userMapper.queryUserListByTenant(tenant.getId());
if (CollectionUtils.isNotEmpty(userList)) {
- putMsg(result, Status.DELETE_TENANT_BY_ID_FAIL_USERS, userList.size());
- return result;
+ throw new ServiceException(Status.DELETE_TENANT_BY_ID_FAIL_USERS, userList.size());
}
// if resource upload startup
@@ -286,11 +305,15 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
public Map<String, Object> queryTenantList(User loginUser) {
Map<String, Object> result = new HashMap<>();
-
- List<Tenant> resourceList = tenantMapper.selectList(null);
+ Set<Integer> ids = resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.TENANT, loginUser.getId(), logger);
+ if (ids.isEmpty()) {
+ result.put(Constants.DATA_LIST, Collections.emptyList());
+ putMsg(result, Status.SUCCESS);
+ return result;
+ }
+ List<Tenant> resourceList = tenantMapper.selectBatchIds(ids);
result.put(Constants.DATA_LIST, resourceList);
putMsg(result, Status.SUCCESS);
-
return result;
}
@@ -304,10 +327,9 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
public Result<Object> verifyTenantCode(String tenantCode) {
Result<Object> result = new Result<>();
if (checkTenantExists(tenantCode)) {
- putMsg(result, Status.OS_TENANT_CODE_EXIST, tenantCode);
- } else {
- putMsg(result, Status.SUCCESS);
+ throw new ServiceException(Status.OS_TENANT_CODE_EXIST, tenantCode);
}
+ putMsg(result, Status.SUCCESS);
return result;
}
@@ -317,8 +339,7 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
* @param tenantCode tenant code
* @return ture if the tenant code exists, otherwise return false
*/
- @Override
- public boolean checkTenantExists(String tenantCode) {
+ private boolean checkTenantExists(String tenantCode) {
Boolean existTenant = tenantMapper.existTenant(tenantCode);
return Boolean.TRUE.equals(existTenant);
}
@@ -339,4 +360,26 @@ public class TenantServiceImpl extends BaseServiceImpl implements TenantService
}
return result;
}
+
+ /**
+ * Make sure tenant with given name exists, and create the tenant if not exists
+ *
+ * ONLY for python gateway server, and should not use this in web ui function
+ *
+ * @param tenantCode tenant code
+ * @param desc The description of tenant object
+ * @param queue The value of queue which current tenant belong
+ * @param queueName The name of queue which current tenant belong
+ * @return Tenant object
+ */
+ @Override
+ public Tenant createTenantIfNotExists(String tenantCode, String desc, String queue, String queueName) {
+ if (checkTenantExists(tenantCode)) {
+ return tenantMapper.queryByTenantCode(tenantCode);
+ }
+
+ tenantCodeValid(tenantCode);
+ Queue newQueue = queueService.createQueueIfNotExists(queue, queueName);
+ return createObjToDB(tenantCode, desc, newQueue.getId());
+ }
}
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/QueueControllerTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/QueueControllerTest.java
index 0b2a4e89a7..c147f84a93 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/QueueControllerTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/controller/QueueControllerTest.java
@@ -42,11 +42,14 @@ import org.springframework.util.MultiValueMap;
public class QueueControllerTest extends AbstractControllerTest {
private static final Logger logger = LoggerFactory.getLogger(QueueControllerTest.class);
- private static final String QUEUE_CREATE_STRING = "queue1";
+ private static final String QUEUE_CREATE_NAME = "queue_create";
+ private static final String QUEUE_MODIFY_NAME = "queue_modify";
+ private static final String QUEUE_NAME_CREATE_NAME = "queue_name_create";
+ private static final String QUEUE_NAME_MODIFY_NAME = "queue_name_modify";
+ private static final String NOT_EXISTS_NAME = "not_exists";
@Test
public void testQueryList() throws Exception {
-
MvcResult mvcResult = mockMvc.perform(get("/queues/list")
.header(SESSION_ID, sessionId))
.andExpect(status().isOk())
@@ -54,13 +57,14 @@ public class QueueControllerTest extends AbstractControllerTest {
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
+
+ Assert.assertNotNull(result);
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
logger.info("query list queue return result:{}", mvcResult.getResponse().getContentAsString());
}
@Test
- public void testQueryQueueListPaging() throws Exception {
-
+ public void testQueryQueueListPagingEmpty() throws Exception {
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("searchVal","");
paramsMap.add("pageNo","1");
@@ -73,17 +77,17 @@ public class QueueControllerTest extends AbstractControllerTest {
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
- Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
- logger.info("query list-paging queue return result:{}", mvcResult.getResponse().getContentAsString());
+ Assert.assertNotNull(result);
+ Assert.assertEquals(Status.SUCCESS.getCode(), result.getCode().intValue());
+ logger.info("query list-paging queue return result:{}", mvcResult.getResponse().getContentAsString());
}
@Test
public void testCreateQueue() throws Exception {
-
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
- paramsMap.add("queue", QUEUE_CREATE_STRING);
- paramsMap.add("queueName","root.queue1");
+ paramsMap.add("queue", QUEUE_CREATE_NAME);
+ paramsMap.add("queueName", QUEUE_NAME_CREATE_NAME);
MvcResult mvcResult = mockMvc.perform(post("/queues")
.header(SESSION_ID, sessionId)
@@ -92,17 +96,18 @@ public class QueueControllerTest extends AbstractControllerTest {
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
+
+ Assert.assertNotNull(result);
Assert.assertEquals(Status.SUCCESS.getCode(), result.getCode().intValue());
logger.info("create queue return result:{}", mvcResult.getResponse().getContentAsString());
}
@Test
public void testUpdateQueue() throws Exception {
-
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
paramsMap.add("id","1");
- paramsMap.add("queue","queue2");
- paramsMap.add("queueName","root.queue2");
+ paramsMap.add("queue", QUEUE_MODIFY_NAME);
+ paramsMap.add("queueName", QUEUE_NAME_MODIFY_NAME);
MvcResult mvcResult = mockMvc.perform(put("/queues/{id}", 1)
.header(SESSION_ID, sessionId)
@@ -111,6 +116,8 @@ public class QueueControllerTest extends AbstractControllerTest {
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
+
+ Assert.assertNotNull(result);
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
logger.info("update queue return result:{}", mvcResult.getResponse().getContentAsString());
}
@@ -120,8 +127,8 @@ public class QueueControllerTest extends AbstractControllerTest {
// queue value exist
MultiValueMap<String, String> paramsMap = new LinkedMultiValueMap<>();
- paramsMap.add("queue",QUEUE_CREATE_STRING);
- paramsMap.add("queueName","queue.name");
+ paramsMap.add("queue", QUEUE_MODIFY_NAME);
+ paramsMap.add("queueName", NOT_EXISTS_NAME);
MvcResult mvcResult = mockMvc.perform(post("/queues/verify")
.header(SESSION_ID, sessionId)
@@ -129,13 +136,32 @@ public class QueueControllerTest extends AbstractControllerTest {
.andExpect(status().isOk())
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
+
Result result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
+
+ Assert.assertNotNull(result);
Assert.assertEquals(Status.QUEUE_VALUE_EXIST.getCode(),result.getCode().intValue());
+ // queue name exist
+ paramsMap.clear();
+ paramsMap.add("queue", NOT_EXISTS_NAME);
+ paramsMap.add("queueName", QUEUE_NAME_CREATE_NAME);
+
+ mvcResult = mockMvc.perform(post("/queues/verify")
+ .header(SESSION_ID, sessionId)
+ .params(paramsMap))
+ .andExpect(status().isOk())
+ .andExpect(content().contentType(MediaType.APPLICATION_JSON))
+ .andReturn();
+ result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
+
+ Assert.assertNotNull(result);
+ Assert.assertEquals(Status.QUEUE_NAME_EXIST.getCode(),result.getCode().intValue());
+
// success
paramsMap.clear();
- paramsMap.add("queue","ait123");
- paramsMap.add("queueName","aitName");
+ paramsMap.add("queue", NOT_EXISTS_NAME);
+ paramsMap.add("queueName", NOT_EXISTS_NAME);
mvcResult = mockMvc.perform(post("/queues/verify")
.header(SESSION_ID, sessionId)
@@ -144,6 +170,8 @@ public class QueueControllerTest extends AbstractControllerTest {
.andExpect(content().contentType(MediaType.APPLICATION_JSON))
.andReturn();
result = JSONUtils.parseObject(mvcResult.getResponse().getContentAsString(), Result.class);
+
+ Assert.assertNotNull(result);
Assert.assertEquals(Status.SUCCESS.getCode(),result.getCode().intValue());
logger.info(mvcResult.getResponse().getContentAsString());
logger.info("verify queue return result:{}", mvcResult.getResponse().getContentAsString());
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
index f3167a560a..b6f1fc6e3f 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/QueueServiceTest.java
@@ -17,7 +17,13 @@
package org.apache.dolphinscheduler.api.service;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_CREATE;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.YARN_QUEUE_UPDATE;
+
import org.apache.dolphinscheduler.api.enums.Status;
+import org.apache.dolphinscheduler.api.exceptions.ServiceException;
+import org.apache.dolphinscheduler.api.permission.ResourcePermissionCheckService;
+import org.apache.dolphinscheduler.api.service.impl.BaseServiceImpl;
import org.apache.dolphinscheduler.api.service.impl.QueueServiceImpl;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
@@ -30,6 +36,7 @@ import org.apache.dolphinscheduler.dao.mapper.UserMapper;
import org.apache.commons.collections.CollectionUtils;
+import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
@@ -38,6 +45,7 @@ import org.junit.After;
import org.junit.Assert;
import org.junit.Before;
import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
@@ -55,7 +63,8 @@ import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
@RunWith(MockitoJUnitRunner.class)
public class QueueServiceTest {
- private static final Logger logger = LoggerFactory.getLogger(QueueServiceTest.class);
+ private static final Logger baseServiceLogger = LoggerFactory.getLogger(BaseServiceImpl.class);
+ private static final Logger queueServiceImplLogger = LoggerFactory.getLogger(QueueServiceImpl.class);
@InjectMocks
private QueueServiceImpl queueService;
@@ -66,7 +75,13 @@ public class QueueServiceTest {
@Mock
private UserMapper userMapper;
- private String queueName = "QueueServiceTest";
+ @Mock
+ private ResourcePermissionCheckService resourcePermissionCheckService;
+
+ private static final String QUEUE = "queue";
+ private static final String QUEUE_NAME = "queueName";
+ private static final String EXISTS = "exists";
+ private static final String NOT_EXISTS = "not_exists";
@Before
public void setUp() {
@@ -78,10 +93,11 @@ public class QueueServiceTest {
@Test
public void testQueryList() {
-
- Mockito.when(queueMapper.selectList(null)).thenReturn(getQueueList());
+ Set<Integer> ids = new HashSet<>();
+ ids.add(1);
+ Mockito.when(resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.QUEUE, getLoginUser().getId(), queueServiceImplLogger)).thenReturn(ids);
+ Mockito.when(queueMapper.selectBatchIds(Mockito.anySet())).thenReturn(getQueueList());
Map<String, Object> result = queueService.queryList(getLoginUser());
- logger.info(result.toString());
List<Queue> queueList = (List<Queue>) result.get(Constants.DATA_LIST);
Assert.assertTrue(CollectionUtils.isNotEmpty(queueList));
@@ -93,92 +109,96 @@ public class QueueServiceTest {
IPage<Queue> page = new Page<>(1, 10);
page.setTotal(1L);
page.setRecords(getQueueList());
- Mockito.when(queueMapper.queryQueuePaging(Mockito.any(Page.class), Mockito.eq(queueName))).thenReturn(page);
- Result result = queueService.queryList(getLoginUser(), queueName, 1, 10);
- logger.info(result.toString());
+ Set<Integer> ids = new HashSet<>();
+ ids.add(1);
+ Mockito.when(resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.QUEUE, getLoginUser().getId(), queueServiceImplLogger)).thenReturn(ids);
+ Mockito.when(queueMapper.queryQueuePaging(Mockito.any(Page.class), Mockito.anyList(), Mockito.eq(QUEUE_NAME))).thenReturn(page);
+ Result result = queueService.queryList(getLoginUser(), QUEUE_NAME, 1, 10);
PageInfo<Queue> pageInfo = (PageInfo<Queue>) result.getData();
Assert.assertTrue(CollectionUtils.isNotEmpty(pageInfo.getTotalList()));
}
@Test
public void testCreateQueue() {
+ Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.QUEUE, getLoginUser().getId(), YARN_QUEUE_CREATE, baseServiceLogger)).thenReturn(true);
+ Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.QUEUE, null, 0, baseServiceLogger)).thenReturn(true);
// queue is null
- Map<String, Object> result = queueService.createQueue(getLoginUser(), null, queueName);
- logger.info(result.toString());
- Assert.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR, result.get(Constants.STATUS));
+ Throwable exception = Assertions.assertThrows(ServiceException.class, () -> queueService.createQueue(getLoginUser(), null, QUEUE_NAME));
+ String formatter = MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), Constants.QUEUE);
+ Assertions.assertEquals(formatter, exception.getMessage());
+
// queueName is null
- result = queueService.createQueue(getLoginUser(), queueName, null);
- logger.info(result.toString());
- Assert.assertEquals(Status.REQUEST_PARAMS_NOT_VALID_ERROR, result.get(Constants.STATUS));
+ exception = Assertions.assertThrows(ServiceException.class, () -> queueService.createQueue(getLoginUser(), QUEUE_NAME, null));
+ formatter = MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), Constants.QUEUE_NAME);
+ Assertions.assertEquals(formatter, exception.getMessage());
+
// correct
- result = queueService.createQueue(getLoginUser(), queueName, queueName);
- logger.info(result.toString());
+ Map<String, Object> result = queueService.createQueue(getLoginUser(), QUEUE_NAME, QUEUE_NAME);
Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
-
}
@Test
public void testUpdateQueue() {
-
- Mockito.when(queueMapper.selectById(1)).thenReturn(getQueue());
- Mockito.when(queueMapper.existQueue("test", null)).thenReturn(true);
- Mockito.when(queueMapper.existQueue(null, "test")).thenReturn(true);
+ Mockito.when(queueMapper.selectById(1)).thenReturn(getQUEUE());
+ Mockito.when(queueMapper.existQueue(EXISTS, null)).thenReturn(true);
+ Mockito.when(queueMapper.existQueue(null, EXISTS)).thenReturn(true);
+ Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.QUEUE, getLoginUser().getId(), YARN_QUEUE_UPDATE, baseServiceLogger)).thenReturn(true);
+ Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.QUEUE, new Object[]{0}, 0, baseServiceLogger)).thenReturn(true);
// not exist
- Map<String, Object> result = queueService.updateQueue(getLoginUser(), 0, "queue", queueName);
- logger.info(result.toString());
- Assert.assertEquals(Status.QUEUE_NOT_EXIST.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
+ Throwable exception = Assertions.assertThrows(ServiceException.class, () -> queueService.updateQueue(getLoginUser(), 0, QUEUE, QUEUE_NAME));
+ String formatter = MessageFormat.format(Status.QUEUE_NOT_EXIST.getMsg(), QUEUE);
+ Assertions.assertEquals(formatter, exception.getMessage());
+
//no need update
- result = queueService.updateQueue(getLoginUser(), 1, queueName, queueName);
- logger.info(result.toString());
- Assert.assertEquals(Status.NEED_NOT_UPDATE_QUEUE.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
+ Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.QUEUE, new Object[]{1}, 0, baseServiceLogger)).thenReturn(true);
+ exception = Assertions.assertThrows(ServiceException.class, () -> queueService.updateQueue(getLoginUser(), 1, QUEUE_NAME, QUEUE_NAME));
+ Assertions.assertEquals(Status.NEED_NOT_UPDATE_QUEUE.getMsg(), exception.getMessage());
+
//queue exist
- result = queueService.updateQueue(getLoginUser(), 1, "test", queueName);
- logger.info(result.toString());
- Assert.assertEquals(Status.QUEUE_VALUE_EXIST.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
+ exception = Assertions.assertThrows(ServiceException.class, () -> queueService.updateQueue(getLoginUser(), 1, EXISTS, QUEUE_NAME));
+ formatter = MessageFormat.format(Status.QUEUE_VALUE_EXIST.getMsg(), EXISTS);
+ Assertions.assertEquals(formatter, exception.getMessage());
+
// queueName exist
- result = queueService.updateQueue(getLoginUser(), 1, "test1", "test");
- logger.info(result.toString());
- Assert.assertEquals(Status.QUEUE_NAME_EXIST.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
+ exception = Assertions.assertThrows(ServiceException.class, () -> queueService.updateQueue(getLoginUser(), 1, NOT_EXISTS, EXISTS));
+ formatter = MessageFormat.format(Status.QUEUE_NAME_EXIST.getMsg(), EXISTS);
+ Assertions.assertEquals(formatter, exception.getMessage());
+
//success
- result = queueService.updateQueue(getLoginUser(), 1, "test1", "test1");
- logger.info(result.toString());
+ Mockito.when(userMapper.existUser(Mockito.anyString())).thenReturn(false);
+ Map<String, Object> result = queueService.updateQueue(getLoginUser(), 1, NOT_EXISTS, NOT_EXISTS);
Assert.assertEquals(Status.SUCCESS.getCode(), ((Status) result.get(Constants.STATUS)).getCode());
-
}
@Test
public void testVerifyQueue() {
-
- Mockito.when(queueMapper.existQueue(queueName, null)).thenReturn(true);
- Mockito.when(queueMapper.existQueue(null, queueName)).thenReturn(true);
-
//queue null
- Result result = queueService.verifyQueue(null, queueName);
- logger.info(result.toString());
- Assert.assertEquals(result.getCode().intValue(), Status.REQUEST_PARAMS_NOT_VALID_ERROR.getCode());
+ Throwable exception = Assertions.assertThrows(ServiceException.class, () -> queueService.verifyQueue(null, QUEUE_NAME));
+ String formatter = MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), Constants.QUEUE);
+ Assertions.assertEquals(formatter, exception.getMessage());
//queueName null
- result = queueService.verifyQueue(queueName, null);
- logger.info(result.toString());
- Assert.assertEquals(result.getCode().intValue(), Status.REQUEST_PARAMS_NOT_VALID_ERROR.getCode());
+ exception = Assertions.assertThrows(ServiceException.class, () -> queueService.verifyQueue(QUEUE_NAME, null));
+ formatter = MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), Constants.QUEUE_NAME);
+ Assertions.assertEquals(formatter, exception.getMessage());
//exist queueName
- result = queueService.verifyQueue(queueName, queueName);
- logger.info(result.toString());
- Assert.assertEquals(result.getCode().intValue(), Status.QUEUE_NAME_EXIST.getCode());
+ Mockito.when(queueMapper.existQueue(EXISTS, null)).thenReturn(true);
+ exception = Assertions.assertThrows(ServiceException.class, () -> queueService.verifyQueue(EXISTS, QUEUE_NAME));
+ formatter = MessageFormat.format(Status.QUEUE_VALUE_EXIST.getMsg(), EXISTS);
+ Assertions.assertEquals(formatter, exception.getMessage());
//exist queue
- result = queueService.verifyQueue(queueName, "test");
- logger.info(result.toString());
- Assert.assertEquals(result.getCode().intValue(), Status.QUEUE_VALUE_EXIST.getCode());
+ Mockito.when(queueMapper.existQueue(null, EXISTS)).thenReturn(true);
+ exception = Assertions.assertThrows(ServiceException.class, () -> queueService.verifyQueue(QUEUE, EXISTS));
+ formatter = MessageFormat.format(Status.QUEUE_NAME_EXIST.getMsg(), EXISTS);
+ Assertions.assertEquals(formatter, exception.getMessage());
// success
- result = queueService.verifyQueue("test", "test");
- logger.info(result.toString());
+ Result<Object> result = queueService.verifyQueue(NOT_EXISTS, NOT_EXISTS);
Assert.assertEquals(result.getCode().intValue(), Status.SUCCESS.getCode());
-
}
/**
@@ -192,26 +212,20 @@ public class QueueServiceTest {
return loginUser;
}
- private List<User> getUserList() {
- List<User> list = new ArrayList<>();
- list.add(getLoginUser());
- return list;
- }
-
/**
* get queue
*/
- private Queue getQueue() {
+ private Queue getQUEUE() {
Queue queue = new Queue();
queue.setId(1);
- queue.setQueue(queueName);
- queue.setQueueName(queueName);
+ queue.setQueue(QUEUE_NAME);
+ queue.setQueueName(QUEUE_NAME);
return queue;
}
private List<Queue> getQueueList() {
List<Queue> queueList = new ArrayList<>();
- queueList.add(getQueue());
+ queueList.add(getQUEUE());
return queueList;
}
diff --git a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
index 5555afc7e5..9ce34afe7a 100644
--- a/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
+++ b/dolphinscheduler-api/src/test/java/org/apache/dolphinscheduler/api/service/TenantServiceTest.java
@@ -17,16 +17,20 @@
package org.apache.dolphinscheduler.api.service;
-import com.baomidou.mybatisplus.core.metadata.IPage;
-import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
-import org.apache.commons.collections.CollectionUtils;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TENANT_CREATE;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TENANT_DELETE;
+import static org.apache.dolphinscheduler.api.constants.ApiFuncIdentificationConstant.TENANT_UPDATE;
+
import org.apache.dolphinscheduler.api.enums.Status;
+import org.apache.dolphinscheduler.api.exceptions.ServiceException;
+import org.apache.dolphinscheduler.api.permission.ResourcePermissionCheckService;
+import org.apache.dolphinscheduler.api.service.impl.BaseServiceImpl;
import org.apache.dolphinscheduler.api.service.impl.TenantServiceImpl;
import org.apache.dolphinscheduler.api.utils.PageInfo;
import org.apache.dolphinscheduler.api.utils.Result;
import org.apache.dolphinscheduler.common.Constants;
+import org.apache.dolphinscheduler.common.enums.AuthorizationType;
import org.apache.dolphinscheduler.common.enums.UserType;
-import org.apache.dolphinscheduler.common.storage.StorageOperate;
import org.apache.dolphinscheduler.common.utils.PropertyUtils;
import org.apache.dolphinscheduler.dao.entity.ProcessDefinition;
import org.apache.dolphinscheduler.dao.entity.ProcessInstance;
@@ -36,8 +40,19 @@ import org.apache.dolphinscheduler.dao.mapper.ProcessDefinitionMapper;
import org.apache.dolphinscheduler.dao.mapper.ProcessInstanceMapper;
import org.apache.dolphinscheduler.dao.mapper.TenantMapper;
import org.apache.dolphinscheduler.dao.mapper.UserMapper;
+
+import org.apache.commons.collections.CollectionUtils;
+
+import java.text.MessageFormat;
+import java.util.ArrayList;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+
import org.junit.Assert;
import org.junit.Test;
+import org.junit.jupiter.api.Assertions;
import org.junit.runner.RunWith;
import org.mockito.InjectMocks;
import org.mockito.Mock;
@@ -47,9 +62,8 @@ import org.powermock.core.classloader.annotations.PrepareForTest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Map;
+import com.baomidou.mybatisplus.core.metadata.IPage;
+import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
/**
* tenant service test
@@ -57,8 +71,8 @@ import java.util.Map;
@RunWith(MockitoJUnitRunner.class)
@PrepareForTest({PropertyUtils.class})
public class TenantServiceTest {
-
- private static final Logger logger = LoggerFactory.getLogger(TenantServiceTest.class);
+ private static final Logger baseServiceLogger = LoggerFactory.getLogger(BaseServiceImpl.class);
+ private static final Logger tenantServiceImplLogger = LoggerFactory.getLogger(TenantServiceImpl.class);
@InjectMocks
private TenantServiceImpl tenantService;
@@ -76,136 +90,143 @@ public class TenantServiceTest {
private UserMapper userMapper;
@Mock
- private StorageOperate storageOperate;
+ private ResourcePermissionCheckService resourcePermissionCheckService;
private static final String tenantCode = "hayden";
+ private static final String tenantDesc = "This is the tenant desc";
@Test
- public void testCreateTenant() {
+ public void testCreateTenant() throws Exception {
User loginUser = getLoginUser();
Mockito.when(tenantMapper.existTenant(tenantCode)).thenReturn(true);
- try {
- //check tenantCode
- Map<String, Object> result =
- tenantService.createTenant(getLoginUser(), "%!1111", 1, "TenantServiceTest");
- logger.info(result.toString());
- Assert.assertEquals(Status.CHECK_OS_TENANT_CODE_ERROR, result.get(Constants.STATUS));
-
- //check exist
- result = tenantService.createTenant(loginUser, tenantCode, 1, "TenantServiceTest");
- logger.info(result.toString());
- Assert.assertEquals(Status.OS_TENANT_CODE_EXIST, result.get(Constants.STATUS));
-
- // success
- result = tenantService.createTenant(loginUser, "test", 1, "TenantServiceTest");
- logger.info(result.toString());
- Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
-
- } catch (Exception e) {
- logger.error("create tenant error", e);
- Assert.fail();
- }
+ Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.TENANT, loginUser.getId(), TENANT_CREATE, baseServiceLogger)).thenReturn(true);
+ Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.TENANT, null, 0, baseServiceLogger)).thenReturn(true);
+ Map<String, Object> result;
+
+ //check exist
+ String emptyTenantCode = "";
+ Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.createTenant(loginUser, emptyTenantCode, 1, tenantDesc));
+ String formatter = MessageFormat.format(Status.REQUEST_PARAMS_NOT_VALID_ERROR.getMsg(), emptyTenantCode);
+ Assertions.assertEquals(formatter, exception.getMessage());
+
+ //check tenant code too long
+ String longStr = "this_is_a_very_long_string_this_is_a_very_long_string_this_is_a_very_long_string_this_is_a_very_long_string";
+ exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.createTenant(loginUser, longStr, 1, tenantDesc));
+ Assert.assertEquals(Status.TENANT_FULL_NAME_TOO_LONG_ERROR.getMsg(), exception.getMessage());
+
+ //check tenant code invalid
+ exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.createTenant(getLoginUser(), "%!1111", 1, tenantDesc));
+ Assert.assertEquals(Status.CHECK_OS_TENANT_CODE_ERROR.getMsg(), exception.getMessage());
+
+ //check exist
+ exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.createTenant(loginUser, tenantCode, 1, tenantDesc));
+ formatter = MessageFormat.format(Status.OS_TENANT_CODE_EXIST.getMsg(), tenantCode);
+ Assert.assertEquals(formatter, exception.getMessage());
+
+ // success
+ result = tenantService.createTenant(loginUser, "test", 1, tenantDesc);
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
+ }
+
+ @Test
+ public void testCreateTenantError() {
+ Mockito.when(tenantMapper.existTenant(tenantCode)).thenReturn(true);
+
+ // tenantCode exist
+ Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.verifyTenantCode(getTenant().getTenantCode()));
+ String expect = MessageFormat.format(Status.OS_TENANT_CODE_EXIST.getMsg(), getTenant().getTenantCode());
+ Assertions.assertEquals(expect, exception.getMessage());
+
+ // success
+ Result result = tenantService.verifyTenantCode("s00000000000l887888885554444sfjdskfjslakslkdf");
+ Assert.assertEquals(Status.SUCCESS.getMsg(), result.getMsg());
}
@Test
@SuppressWarnings("unchecked")
public void testQueryTenantListPage() {
-
IPage<Tenant> page = new Page<>(1, 10);
page.setRecords(getList());
page.setTotal(1L);
- Mockito.when(tenantMapper.queryTenantPaging(Mockito.any(Page.class), Mockito.eq("TenantServiceTest")))
+ Set<Integer> ids = new HashSet<>();
+ ids.add(1);
+ Mockito.when(resourcePermissionCheckService.userOwnedResourceIdsAcquisition(AuthorizationType.TENANT, getLoginUser().getId(), tenantServiceImplLogger)).thenReturn(ids);
+ Mockito.when(tenantMapper.queryTenantPaging(Mockito.any(Page.class), Mockito.anyList(), Mockito.eq(tenantDesc)))
.thenReturn(page);
- Result result = tenantService.queryTenantList(getLoginUser(), "TenantServiceTest", 1, 10);
- logger.info(result.toString());
+ Result result = tenantService.queryTenantList(getLoginUser(), tenantDesc, 1, 10);
PageInfo<Tenant> pageInfo = (PageInfo<Tenant>) result.getData();
Assert.assertTrue(CollectionUtils.isNotEmpty(pageInfo.getTotalList()));
}
@Test
- public void testUpdateTenant() {
-
+ public void testUpdateTenant() throws Exception {
Mockito.when(tenantMapper.queryById(1)).thenReturn(getTenant());
- try {
- // id not exist
- Map<String, Object> result =
- tenantService.updateTenant(getLoginUser(), 912222, tenantCode, 1, "desc");
- logger.info(result.toString());
- // success
- Assert.assertEquals(Status.TENANT_NOT_EXIST, result.get(Constants.STATUS));
- result = tenantService.updateTenant(getLoginUser(), 1, tenantCode, 1, "desc");
- logger.info(result.toString());
- Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
- } catch (Exception e) {
- logger.error("update tenant error", e);
- Assert.fail();
- }
+ Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.TENANT, getLoginUser().getId(), TENANT_UPDATE, baseServiceLogger)).thenReturn(true);
+ Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.TENANT, null, 0, baseServiceLogger)).thenReturn(true);
+
+ Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.updateTenant(getLoginUser(), 912222, tenantCode, 1, tenantDesc));
+ Assertions.assertEquals(Status.TENANT_NOT_EXIST.getMsg(), exception.getMessage());
+ Map<String, Object> result = tenantService.updateTenant(getLoginUser(), 1, tenantCode, 1, tenantDesc);
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@Test
- public void testDeleteTenantById() {
-
+ public void testDeleteTenantById() throws Exception {
+ Mockito.when(resourcePermissionCheckService.operationPermissionCheck(AuthorizationType.TENANT, getLoginUser().getId(), TENANT_DELETE, baseServiceLogger)).thenReturn(true);
+ Mockito.when(resourcePermissionCheckService.resourcePermissionCheck(AuthorizationType.TENANT, null, 0, baseServiceLogger)).thenReturn(true);
Mockito.when(tenantMapper.queryById(1)).thenReturn(getTenant());
Mockito.when(processInstanceMapper.queryByTenantIdAndStatus(1, Constants.NOT_TERMINATED_STATES))
.thenReturn(getInstanceList());
Mockito.when(processDefinitionMapper.queryDefinitionListByTenant(2)).thenReturn(getDefinitionsList());
Mockito.when(userMapper.queryUserListByTenant(3)).thenReturn(getUserList());
- try {
- //TENANT_NOT_EXIST
- Map<String, Object> result = tenantService.deleteTenantById(getLoginUser(), 12);
- logger.info(result.toString());
- Assert.assertEquals(Status.TENANT_NOT_EXIST, result.get(Constants.STATUS));
-
- //DELETE_TENANT_BY_ID_FAIL
- result = tenantService.deleteTenantById(getLoginUser(), 1);
- logger.info(result.toString());
- Assert.assertEquals(Status.DELETE_TENANT_BY_ID_FAIL, result.get(Constants.STATUS));
-
- //DELETE_TENANT_BY_ID_FAIL_DEFINES
- Mockito.when(tenantMapper.queryById(2)).thenReturn(getTenant(2));
- result = tenantService.deleteTenantById(getLoginUser(), 2);
- logger.info(result.toString());
- Assert.assertEquals(Status.DELETE_TENANT_BY_ID_FAIL_DEFINES, result.get(Constants.STATUS));
-
- //DELETE_TENANT_BY_ID_FAIL_USERS
- Mockito.when(tenantMapper.queryById(3)).thenReturn(getTenant(3));
- result = tenantService.deleteTenantById(getLoginUser(), 3);
- logger.info(result.toString());
- Assert.assertEquals(Status.DELETE_TENANT_BY_ID_FAIL_USERS, result.get(Constants.STATUS));
-
- // success
- Mockito.when(tenantMapper.queryById(4)).thenReturn(getTenant(4));
- result = tenantService.deleteTenantById(getLoginUser(), 4);
- logger.info(result.toString());
- Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
- } catch (Exception e) {
- logger.error("delete tenant error", e);
- Assert.fail();
- }
+ //TENANT_NOT_EXIST
+ Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.deleteTenantById(getLoginUser(), 12));
+ Assertions.assertEquals(Status.TENANT_NOT_EXIST.getMsg(), exception.getMessage());
+
+ //DELETE_TENANT_BY_ID_FAIL
+ exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.deleteTenantById(getLoginUser(), 1));
+ String prefix = Status.DELETE_TENANT_BY_ID_FAIL.getMsg().substring(1, 5);
+ Assertions.assertTrue(exception.getMessage().contains(prefix));
+
+ //DELETE_TENANT_BY_ID_FAIL_DEFINES
+ Mockito.when(tenantMapper.queryById(2)).thenReturn(getTenant(2));
+ exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.deleteTenantById(getLoginUser(), 2));
+ prefix = Status.DELETE_TENANT_BY_ID_FAIL_DEFINES.getMsg().substring(1, 5);
+ Assertions.assertTrue(exception.getMessage().contains(prefix));
+
+ //DELETE_TENANT_BY_ID_FAIL_USERS
+ Mockito.when(tenantMapper.queryById(3)).thenReturn(getTenant(3));
+ exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.deleteTenantById(getLoginUser(), 3));
+ prefix = Status.DELETE_TENANT_BY_ID_FAIL_USERS.getMsg().substring(1, 5);
+ Assertions.assertTrue(exception.getMessage().contains(prefix));
+
+ // success
+ Mockito.when(tenantMapper.queryById(4)).thenReturn(getTenant(4));
+ Map<String, Object> result = tenantService.deleteTenantById(getLoginUser(), 4);
+ Assert.assertEquals(Status.SUCCESS, result.get(Constants.STATUS));
}
@Test
public void testVerifyTenantCode() {
-
Mockito.when(tenantMapper.existTenant(tenantCode)).thenReturn(true);
- // tenantCode not exist
+ // tenantCode exist
+ Throwable exception = Assertions.assertThrows(ServiceException.class, () -> tenantService.verifyTenantCode(getTenant().getTenantCode()));
+ String expect = MessageFormat.format(Status.OS_TENANT_CODE_EXIST.getMsg(), getTenant().getTenantCode());
+ Assertions.assertEquals(expect, exception.getMessage());
+
+ // success
Result result = tenantService.verifyTenantCode("s00000000000l887888885554444sfjdskfjslakslkdf");
- logger.info(result.toString());
Assert.assertEquals(Status.SUCCESS.getMsg(), result.getMsg());
- // tenantCode exist
- result = tenantService.verifyTenantCode(getTenant().getTenantCode());
- Assert.assertEquals(Status.OS_TENANT_CODE_EXIST.getCode(), result.getCode().intValue());
}
/**
* get user
*/
private User getLoginUser() {
-
User loginUser = new User();
loginUser.setUserType(UserType.ADMIN_USER);
return loginUser;
diff --git a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/QueueMapper.java b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/QueueMapper.java
index e48607007d..86ffd0314e 100644
--- a/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/QueueMapper.java
+++ b/dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/mapper/QueueMapper.java
@@ -55,9 +55,10 @@ public interface QueueMapper extends BaseMapper<Queue> {
Boolean existQueue(@Param("queue") String queue, @Param("queueName") String queueName);
/**
- * query queue by queue name
+ * query simple queue object by queue name and queue
+ * @param queue queue
* @param queueName queueName
- * @return queue list
+ * @return queue object
*/
- List<Queue> queryQueueName(@Param("queueName") String queueName);
+ Queue queryQueueName(@Param("queue") String queue, @Param("queueName") String queueName);
}
diff --git a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/QueueMapper.xml b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/QueueMapper.xml
index cf381e494a..fe5a48b91b 100644
--- a/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/QueueMapper.xml
+++ b/dolphinscheduler-dao/src/main/resources/org/apache/dolphinscheduler/dao/mapper/QueueMapper.xml
@@ -58,9 +58,13 @@
select
<include refid="baseSql"/>
from t_ds_queue
- where 1 = 1
- <if test="queueName != null and queueName != ''">
- and queue_name =#{queueName}
- </if>
+ <where>
+ <if test="queue != null and queue != ''">
+ and queue =#{queue}
+ </if>
+ <if test="queueName != null and queueName != ''">
+ and queue_name =#{queueName}
+ </if>
+ </where>
</select>
</mapper>