@ -16,9 +16,15 @@
* /
* /
package org.apache.dolphinscheduler.api.service ;
package org.apache.dolphinscheduler.api.service ;
import com.alibaba.fastjson.JSON ;
import com.alibaba.fastjson.serializer.SerializerFeature ;
import com.baomidou.mybatisplus.core.metadata.IPage ;
import com.baomidou.mybatisplus.core.metadata.IPage ;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page ;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page ;
import org.apache.commons.collections.BeanMap ;
import org.apache.commons.collections.BeanMap ;
import org.apache.dolphinscheduler.api.dto.resources.ResourceComponent ;
import org.apache.dolphinscheduler.api.dto.resources.filter.ResourceFilter ;
import org.apache.dolphinscheduler.api.dto.resources.visitor.ResourceTreeVisitor ;
import org.apache.dolphinscheduler.api.dto.resources.visitor.Visitor ;
import org.apache.dolphinscheduler.api.enums.Status ;
import org.apache.dolphinscheduler.api.enums.Status ;
import org.apache.dolphinscheduler.api.utils.PageInfo ;
import org.apache.dolphinscheduler.api.utils.PageInfo ;
import org.apache.dolphinscheduler.api.utils.Result ;
import org.apache.dolphinscheduler.api.utils.Result ;
@ -39,6 +45,7 @@ import org.springframework.web.multipart.MultipartFile;
import java.text.MessageFormat ;
import java.text.MessageFormat ;
import java.util.* ;
import java.util.* ;
import java.util.stream.Collectors ;
import static org.apache.dolphinscheduler.common.Constants.* ;
import static org.apache.dolphinscheduler.common.Constants.* ;
@ -65,6 +72,82 @@ public class ResourcesService extends BaseService {
@Autowired
@Autowired
private ResourceUserMapper resourceUserMapper ;
private ResourceUserMapper resourceUserMapper ;
@Autowired
private ProcessDefinitionMapper processDefinitionMapper ;
/ * *
* create directory
*
* @param loginUser login user
* @param name alias
* @param description description
* @param type type
* @param pid parent id
* @param currentDir current directory
* @return create directory result
* /
@Transactional ( rollbackFor = Exception . class )
public Result createDirectory ( User loginUser ,
String name ,
String description ,
ResourceType type ,
int pid ,
String currentDir ) {
Result result = new Result ( ) ;
// if hdfs not startup
if ( ! PropertyUtils . getResUploadStartupState ( ) ) {
logger . error ( "resource upload startup state: {}" , PropertyUtils . getResUploadStartupState ( ) ) ;
putMsg ( result , Status . HDFS_NOT_STARTUP ) ;
return result ;
}
String fullName = currentDir . equals ( "/" ) ? String . format ( "%s%s" , currentDir , name ) : String . format ( "%s/%s" , currentDir , name ) ;
if ( pid ! = - 1 ) {
Resource parentResource = resourcesMapper . selectById ( pid ) ;
if ( parentResource = = null ) {
putMsg ( result , Status . PARENT_RESOURCE_NOT_EXIST ) ;
return result ;
}
if ( ! hasPerm ( loginUser , parentResource . getUserId ( ) ) ) {
putMsg ( result , Status . USER_NO_OPERATION_PERM ) ;
return result ;
}
}
if ( checkResourceExists ( fullName , 0 , type . ordinal ( ) ) ) {
logger . error ( "resource directory {} has exist, can't recreate" , fullName ) ;
putMsg ( result , Status . RESOURCE_EXIST ) ;
return result ;
}
Date now = new Date ( ) ;
Resource resource = new Resource ( pid , name , fullName , true , description , name , loginUser . getId ( ) , type , 0 , now , now ) ;
try {
resourcesMapper . insert ( resource ) ;
putMsg ( result , Status . SUCCESS ) ;
Map < Object , Object > dataMap = new BeanMap ( resource ) ;
Map < String , Object > resultMap = new HashMap < String , Object > ( ) ;
for ( Map . Entry < Object , Object > entry : dataMap . entrySet ( ) ) {
if ( ! "class" . equalsIgnoreCase ( entry . getKey ( ) . toString ( ) ) ) {
resultMap . put ( entry . getKey ( ) . toString ( ) , entry . getValue ( ) ) ;
}
}
result . setData ( resultMap ) ;
} catch ( Exception e ) {
logger . error ( "resource already exists, can't recreate " , e ) ;
throw new RuntimeException ( "resource already exists, can't recreate" ) ;
}
//create directory in hdfs
createDirecotry ( loginUser , fullName , type , result ) ;
return result ;
}
/ * *
/ * *
* create resource
* create resource
*
*
@ -73,6 +156,8 @@ public class ResourcesService extends BaseService {
* @param desc description
* @param desc description
* @param file file
* @param file file
* @param type type
* @param type type
* @param pid parent id
* @param currentDir current directory
* @return create result code
* @return create result code
* /
* /
@Transactional ( rollbackFor = Exception . class )
@Transactional ( rollbackFor = Exception . class )
@ -80,7 +165,9 @@ public class ResourcesService extends BaseService {
String name ,
String name ,
String desc ,
String desc ,
ResourceType type ,
ResourceType type ,
MultipartFile file ) {
MultipartFile file ,
int pid ,
String currentDir ) {
Result result = new Result ( ) ;
Result result = new Result ( ) ;
// if hdfs not startup
// if hdfs not startup
@ -123,7 +210,8 @@ public class ResourcesService extends BaseService {
}
}
// check resoure name exists
// check resoure name exists
if ( checkResourceExists ( name , 0 , type . ordinal ( ) ) ) {
String fullName = currentDir . equals ( "/" ) ? String . format ( "%s%s" , currentDir , name ) : String . format ( "%s/%s" , currentDir , name ) ;
if ( checkResourceExists ( fullName , 0 , type . ordinal ( ) ) ) {
logger . error ( "resource {} has exist, can't recreate" , name ) ;
logger . error ( "resource {} has exist, can't recreate" , name ) ;
putMsg ( result , Status . RESOURCE_EXIST ) ;
putMsg ( result , Status . RESOURCE_EXIST ) ;
return result ;
return result ;
@ -131,7 +219,9 @@ public class ResourcesService extends BaseService {
Date now = new Date ( ) ;
Date now = new Date ( ) ;
Resource resource = new Resource ( name , file . getOriginalFilename ( ) , desc , loginUser . getId ( ) , type , file . getSize ( ) , now , now ) ;
Resource resource = new Resource ( pid , name , fullName , false , desc , file . getOriginalFilename ( ) , loginUser . getId ( ) , type , file . getSize ( ) , now , now ) ;
try {
try {
resourcesMapper . insert ( resource ) ;
resourcesMapper . insert ( resource ) ;
@ -151,7 +241,7 @@ public class ResourcesService extends BaseService {
}
}
// fail upload
// fail upload
if ( ! upload ( loginUser , n ame, file , type ) ) {
if ( ! upload ( loginUser , fullN ame, file , type ) ) {
logger . error ( "upload resource: {} file: {} failed." , name , file . getOriginalFilename ( ) ) ;
logger . error ( "upload resource: {} file: {} failed." , name , file . getOriginalFilename ( ) ) ;
putMsg ( result , Status . HDFS_OPERATION_ERROR ) ;
putMsg ( result , Status . HDFS_OPERATION_ERROR ) ;
throw new RuntimeException ( String . format ( "upload resource: %s file: %s failed." , name , file . getOriginalFilename ( ) ) ) ;
throw new RuntimeException ( String . format ( "upload resource: %s file: %s failed." , name , file . getOriginalFilename ( ) ) ) ;
@ -162,26 +252,28 @@ public class ResourcesService extends BaseService {
/ * *
/ * *
* check resource is exists
* check resource is exists
*
*
* @param alias alias
* @param fullName fullName
* @param userId user id
* @param userId user id
* @param type type
* @param type type
* @return true if resource exists
* @return true if resource exists
* /
* /
private boolean checkResourceExists ( String alias , int userId , int type ) {
private boolean checkResourceExists ( String fullName , int userId , int type ) {
List < Resource > resources = resourcesMapper . queryResourceList ( alias , userId , type ) ;
return CollectionUtils . isNotEmpty ( resources ) ;
}
List < Resource > resources = resourcesMapper . queryResourceList ( fullName , userId , type ) ;
if ( resources ! = null & & resources . size ( ) > 0 ) {
return true ;
}
return false ;
}
/ * *
/ * *
* update resource
* update resource
*
* @param loginUser login user
* @param loginUser login user
* @param name alias
* @param resourceId resource id
* @param resourceId resource id
* @param type resource typ e
* @param name nam e
* @param desc description
* @param desc description
* @param type resource type
* @return update result code
* @return update result code
* /
* /
@Transactional ( rollbackFor = Exception . class )
@Transactional ( rollbackFor = Exception . class )
@ -216,7 +308,10 @@ public class ResourcesService extends BaseService {
}
}
//check resource aleady exists
//check resource aleady exists
if ( ! resource . getAlias ( ) . equals ( name ) & & checkResourceExists ( name , 0 , type . ordinal ( ) ) ) {
String originFullName = resource . getFullName ( ) ;
String fullName = String . format ( "%s%s" , originFullName . substring ( 0 , originFullName . lastIndexOf ( "/" ) + 1 ) , name ) ;
if ( ! resource . getAlias ( ) . equals ( name ) & & checkResourceExists ( fullName , 0 , type . ordinal ( ) ) ) {
logger . error ( "resource {} already exists, can't recreate" , name ) ;
logger . error ( "resource {} already exists, can't recreate" , name ) ;
putMsg ( result , Status . RESOURCE_EXIST ) ;
putMsg ( result , Status . RESOURCE_EXIST ) ;
return result ;
return result ;
@ -227,25 +322,41 @@ public class ResourcesService extends BaseService {
if ( StringUtils . isEmpty ( tenantCode ) ) {
if ( StringUtils . isEmpty ( tenantCode ) ) {
return result ;
return result ;
}
}
String nameWithSuffix = name ;
//get the file suffix
String originResourceName = resource . getAlias ( ) ;
String originResourceName = resource . getAlias ( ) ;
String suffix = originResourceName . substring ( originResourceName . lastIndexOf ( '.' ) ) ;
if ( ! resource . isDirectory ( ) ) {
//get the file suffix
String suffix = originResourceName . substring ( originResourceName . lastIndexOf ( "." ) ) ;
//if the name without suffix then add it ,else use the origin name
//if the name without suffix then add it ,else use the origin name
String nameWithSuffix = name ;
if ( ! name . endsWith ( suffix ) ) {
if ( ! name . endsWith ( suffix ) ) {
nameWithSuffix = nameWithSuffix + suffix ;
nameWithSuffix = nameWithSuffix + suffix ;
}
}
}
// updateResource data
// updateResource data
List < Integer > childrenResource = listAllChildren ( resource ) ;
String oldFullName = resource . getFullName ( ) ;
Date now = new Date ( ) ;
Date now = new Date ( ) ;
resource . setAlias ( nameWithSuffix ) ;
resource . setAlias ( nameWithSuffix ) ;
resource . setFullName ( fullName ) ;
resource . setDescription ( desc ) ;
resource . setDescription ( desc ) ;
resource . setUpdateTime ( now ) ;
resource . setUpdateTime ( now ) ;
try {
try {
resourcesMapper . updateById ( resource ) ;
resourcesMapper . updateById ( resource ) ;
if ( resource . isDirectory ( ) & & CollectionUtils . isNotEmpty ( childrenResource ) ) {
List < Resource > childResourceList = new ArrayList < > ( ) ;
List < Resource > resourceList = resourcesMapper . listResourceByIds ( childrenResource . toArray ( new Integer [ childrenResource . size ( ) ] ) ) ;
childResourceList = resourceList . stream ( ) . map ( t - > {
t . setFullName ( t . getFullName ( ) . replaceFirst ( oldFullName , fullName ) ) ;
t . setUpdateTime ( now ) ;
return t ;
} ) . collect ( Collectors . toList ( ) ) ;
resourcesMapper . batchUpdateResource ( childResourceList ) ;
}
putMsg ( result , Status . SUCCESS ) ;
putMsg ( result , Status . SUCCESS ) ;
Map < Object , Object > dataMap = new BeanMap ( resource ) ;
Map < Object , Object > dataMap = new BeanMap ( resource ) ;
@ -267,15 +378,9 @@ public class ResourcesService extends BaseService {
// get file hdfs path
// get file hdfs path
// delete hdfs file by type
// delete hdfs file by type
String originHdfsFileName = "" ;
String originHdfsFileName = HadoopUtils . getHdfsFileName ( resource . getType ( ) , tenantCode , originFullName ) ;
String destHdfsFileName = "" ;
String destHdfsFileName = HadoopUtils . getHdfsFileName ( resource . getType ( ) , tenantCode , fullName ) ;
if ( resource . getType ( ) . equals ( ResourceType . FILE ) ) {
originHdfsFileName = HadoopUtils . getHdfsFilename ( tenantCode , originResourceName ) ;
destHdfsFileName = HadoopUtils . getHdfsFilename ( tenantCode , name ) ;
} else if ( resource . getType ( ) . equals ( ResourceType . UDF ) ) {
originHdfsFileName = HadoopUtils . getHdfsUdfFilename ( tenantCode , originResourceName ) ;
destHdfsFileName = HadoopUtils . getHdfsUdfFilename ( tenantCode , name ) ;
}
try {
try {
if ( HadoopUtils . getInstance ( ) . exists ( originHdfsFileName ) ) {
if ( HadoopUtils . getInstance ( ) . exists ( originHdfsFileName ) ) {
logger . info ( "hdfs copy {} -> {}" , originHdfsFileName , destHdfsFileName ) ;
logger . info ( "hdfs copy {} -> {}" , originHdfsFileName , destHdfsFileName ) ;
@ -303,7 +408,7 @@ public class ResourcesService extends BaseService {
* @param pageSize page size
* @param pageSize page size
* @return resource list page
* @return resource list page
* /
* /
public Map < String , Object > queryResourceListPaging ( User loginUser , ResourceType type , String searchVal , Integer pageNo , Integer pageSize ) {
public Map < String , Object > queryResourceListPaging ( User loginUser , int direcotryId , ResourceType type , String searchVal , Integer pageNo , Integer pageSize ) {
HashMap < String , Object > result = new HashMap < > ( 5 ) ;
HashMap < String , Object > result = new HashMap < > ( 5 ) ;
Page < Resource > page = new Page ( pageNo , pageSize ) ;
Page < Resource > page = new Page ( pageNo , pageSize ) ;
@ -312,7 +417,7 @@ public class ResourcesService extends BaseService {
userId = 0 ;
userId = 0 ;
}
}
IPage < Resource > resourceIPage = resourcesMapper . queryResourcePaging ( page ,
IPage < Resource > resourceIPage = resourcesMapper . queryResourcePaging ( page ,
userId , type . ordinal ( ) , searchVal ) ;
userId , direcotryId , type . ordinal ( ) , searchVal ) ;
PageInfo pageInfo = new PageInfo < Resource > ( pageNo , pageSize ) ;
PageInfo pageInfo = new PageInfo < Resource > ( pageNo , pageSize ) ;
pageInfo . setTotalCount ( ( int ) resourceIPage . getTotal ( ) ) ;
pageInfo . setTotalCount ( ( int ) resourceIPage . getTotal ( ) ) ;
pageInfo . setLists ( resourceIPage . getRecords ( ) ) ;
pageInfo . setLists ( resourceIPage . getRecords ( ) ) ;
@ -321,17 +426,46 @@ public class ResourcesService extends BaseService {
return result ;
return result ;
}
}
/ * *
* create direcoty
* @param loginUser login user
* @param fullName full name
* @param type resource type
* @param result Result
* /
private void createDirecotry ( User loginUser , String fullName , ResourceType type , Result result ) {
// query tenant
String tenantCode = tenantMapper . queryById ( loginUser . getTenantId ( ) ) . getTenantCode ( ) ;
String directoryName = HadoopUtils . getHdfsFileName ( type , tenantCode , fullName ) ;
String resourceRootPath = HadoopUtils . getHdfsDir ( type , tenantCode ) ;
try {
if ( ! HadoopUtils . getInstance ( ) . exists ( resourceRootPath ) ) {
createTenantDirIfNotExists ( tenantCode ) ;
}
if ( ! HadoopUtils . getInstance ( ) . mkdir ( directoryName ) ) {
logger . error ( "create resource directory {} of hdfs failed" , directoryName ) ;
putMsg ( result , Status . HDFS_OPERATION_ERROR ) ;
throw new RuntimeException ( String . format ( "create resource directory: %s failed." , directoryName ) ) ;
}
} catch ( Exception e ) {
logger . error ( "create resource directory {} of hdfs failed" , directoryName ) ;
putMsg ( result , Status . HDFS_OPERATION_ERROR ) ;
throw new RuntimeException ( String . format ( "create resource directory: %s failed." , directoryName ) ) ;
}
}
/ * *
/ * *
* upload file to hdfs
* upload file to hdfs
*
*
* @param loginUser
* @param loginUser login user
* @param name
* @param fullName full name
* @param file
* @param file file
* /
* /
private boolean upload ( User loginUser , String name , MultipartFile file , ResourceType type ) {
private boolean upload ( User loginUser , String fullN ame, MultipartFile file , ResourceType type ) {
// save to local
// save to local
String fileSuffix = FileUtils . suffix ( file . getOriginalFilename ( ) ) ;
String fileSuffix = FileUtils . suffix ( file . getOriginalFilename ( ) ) ;
String nameSuffix = FileUtils . suffix ( name ) ;
String nameSuffix = FileUtils . suffix ( fullN ame) ;
// determine file suffix
// determine file suffix
if ( ! ( StringUtils . isNotEmpty ( fileSuffix ) & & fileSuffix . equalsIgnoreCase ( nameSuffix ) ) ) {
if ( ! ( StringUtils . isNotEmpty ( fileSuffix ) & & fileSuffix . equalsIgnoreCase ( nameSuffix ) ) ) {
@ -344,15 +478,8 @@ public class ResourcesService extends BaseService {
// save file to hdfs, and delete original file
// save file to hdfs, and delete original file
String hdfsFilename = "" ;
String hdfsFilename = HadoopUtils . getHdfsFileName ( type , tenantCode , fullName ) ;
String resourcePath = "" ;
String resourcePath = HadoopUtils . getHdfsDir ( type , tenantCode ) ;
if ( type . equals ( ResourceType . FILE ) ) {
hdfsFilename = HadoopUtils . getHdfsFilename ( tenantCode , name ) ;
resourcePath = HadoopUtils . getHdfsResDir ( tenantCode ) ;
} else if ( type . equals ( ResourceType . UDF ) ) {
hdfsFilename = HadoopUtils . getHdfsUdfFilename ( tenantCode , name ) ;
resourcePath = HadoopUtils . getHdfsUdfDir ( tenantCode ) ;
}
try {
try {
// if tenant dir not exists
// if tenant dir not exists
if ( ! HadoopUtils . getInstance ( ) . exists ( resourcePath ) ) {
if ( ! HadoopUtils . getInstance ( ) . exists ( resourcePath ) ) {
@ -377,13 +504,59 @@ public class ResourcesService extends BaseService {
public Map < String , Object > queryResourceList ( User loginUser , ResourceType type ) {
public Map < String , Object > queryResourceList ( User loginUser , ResourceType type ) {
Map < String , Object > result = new HashMap < > ( 5 ) ;
Map < String , Object > result = new HashMap < > ( 5 ) ;
List < Resource > resourceList ;
Set < Resource > allResourceList = getAllResources ( loginUser , type ) ;
Visitor resourceTreeVisitor = new ResourceTreeVisitor ( new ArrayList < > ( allResourceList ) ) ;
//JSONArray jsonArray = JSON.parseArray(JSON.toJSONString(resourceTreeVisitor.visit().getChildren(), SerializerFeature.SortField));
result . put ( Constants . DATA_LIST , resourceTreeVisitor . visit ( ) . getChildren ( ) ) ;
putMsg ( result , Status . SUCCESS ) ;
return result ;
}
/ * *
* get all resources
* @param loginUser login user
* @return all resource set
* /
private Set < Resource > getAllResources ( User loginUser , ResourceType type ) {
int userId = loginUser . getId ( ) ;
int userId = loginUser . getId ( ) ;
boolean listChildren = true ;
if ( isAdmin ( loginUser ) ) {
if ( isAdmin ( loginUser ) ) {
userId = 0 ;
userId = 0 ;
listChildren = false ;
}
List < Resource > resourceList = resourcesMapper . queryResourceListAuthored ( userId , type . ordinal ( ) ) ;
Set < Resource > allResourceList = new HashSet < > ( resourceList ) ;
if ( listChildren ) {
Set < Integer > authorizedIds = new HashSet < > ( ) ;
List < Resource > authorizedDirecoty = resourceList . stream ( ) . filter ( t - > t . getUserId ( ) ! = loginUser . getId ( ) & & t . isDirectory ( ) ) . collect ( Collectors . toList ( ) ) ;
if ( CollectionUtils . isNotEmpty ( authorizedDirecoty ) ) {
for ( Resource resource : authorizedDirecoty ) {
authorizedIds . addAll ( listAllChildren ( resource ) ) ;
}
List < Resource > childrenResources = resourcesMapper . listResourceByIds ( authorizedIds . toArray ( new Integer [ authorizedIds . size ( ) ] ) ) ;
allResourceList . addAll ( childrenResources ) ;
}
}
resourceList = resourcesMapper . queryResourceListAuthored ( userId , type . ordinal ( ) ) ;
}
result . put ( Constants . DATA_LIST , resourceList ) ;
return allResourceList ;
}
/ * *
* query resource list
*
* @param loginUser login user
* @param type resource type
* @return resource list
* /
public Map < String , Object > queryResourceJarList ( User loginUser , ResourceType type ) {
Map < String , Object > result = new HashMap < > ( 5 ) ;
Set < Resource > allResourceList = getAllResources ( loginUser , type ) ;
List < Resource > resources = new ResourceFilter ( ".jar" , new ArrayList < > ( allResourceList ) ) . filter ( ) ;
Visitor resourceTreeVisitor = new ResourceTreeVisitor ( resources ) ;
result . put ( Constants . DATA_LIST , resourceTreeVisitor . visit ( ) . getChildren ( ) ) ;
putMsg ( result , Status . SUCCESS ) ;
putMsg ( result , Status . SUCCESS ) ;
return result ;
return result ;
@ -419,23 +592,51 @@ public class ResourcesService extends BaseService {
putMsg ( result , Status . USER_NO_OPERATION_PERM ) ;
putMsg ( result , Status . USER_NO_OPERATION_PERM ) ;
return result ;
return result ;
}
}
//if resource type is UDF,need check whether it is bound by UDF functon
if ( resource . getType ( ) = = ( ResourceType . UDF ) ) {
List < UdfFunc > udfFuncs = udfFunctionMapper . listUdfByResourceId ( new int [ ] { resourceId } ) ;
if ( CollectionUtils . isNotEmpty ( udfFuncs ) ) {
logger . error ( "can't be deleted,because it is bound by UDF functions:{}" , udfFuncs . toString ( ) ) ;
putMsg ( result , Status . UDF_RESOURCE_IS_BOUND , udfFuncs . get ( 0 ) . getFuncName ( ) ) ;
return result ;
}
}
Tenant tenant = tenantMapper . queryById ( loginUser . getTenantId ( ) ) ;
String tenantCode = getTenantCode ( resource . getUserId ( ) , result ) ;
if ( tenant = = null ) {
if ( StringUtils . isEmpty ( tenantCode ) ) {
putMsg ( result , Status . TENANT_NOT_EXIST ) ;
return result ;
return result ;
}
}
String hdfsFilename = "" ;
// delete hdfs file by type
// get all resource id of process definitions those is released
String tenantCode = tenant . getTenantCode ( ) ;
Map < Integer , Set < Integer > > resourceProcessMap = getResourceProcessMap ( ) ;
hdfsFilename = getHdfsFileName ( resource , tenantCode , hdfsFilename ) ;
Set < Integer > resourceIdSet = resourceProcessMap . keySet ( ) ;
// get all children of the resource
List < Integer > allChildren = listAllChildren ( resource ) ;
if ( resourceIdSet . contains ( resource . getPid ( ) ) ) {
logger . error ( "can't be deleted,because it is used of process definition" ) ;
putMsg ( result , Status . RESOURCE_IS_USED ) ;
return result ;
}
resourceIdSet . retainAll ( allChildren ) ;
if ( CollectionUtils . isNotEmpty ( resourceIdSet ) ) {
logger . error ( "can't be deleted,because it is used of process definition" ) ;
for ( Integer resId : resourceIdSet ) {
logger . error ( "resource id:{} is used of process definition {}" , resId , resourceProcessMap . get ( resId ) ) ;
}
putMsg ( result , Status . RESOURCE_IS_USED ) ;
return result ;
}
// get hdfs file by type
String hdfsFilename = HadoopUtils . getHdfsFileName ( resource . getType ( ) , tenantCode , resource . getFullName ( ) ) ;
//delete data in database
//delete data in database
resourcesMapper . deleteById ( resourceId ) ;
resourcesMapper . deleteIds ( allChildren . toArray ( new Integer [ allChildren . size ( ) ] ) ) ;
resourceUserMapper . deleteResourceUser ( 0 , resourceId ) ;
resourceUserMapper . deleteResourceUser ( 0 , resourceId ) ;
//delete file on hdfs
//delete file on hdfs
HadoopUtils . getInstance ( ) . delete ( hdfsFilename , false ) ;
HadoopUtils . getInstance ( ) . delete ( hdfsFilename , tru e) ;
putMsg ( result , Status . SUCCESS ) ;
putMsg ( result , Status . SUCCESS ) ;
return result ;
return result ;
@ -444,15 +645,15 @@ public class ResourcesService extends BaseService {
/ * *
/ * *
* verify resource by name and type
* verify resource by name and type
* @param loginUser login user
* @param loginUser login user
* @param name resource alias
* @param fullName resource full name
* @param type resource type
* @param type resource type
* @return true if the resource name not exists , otherwise return false
* @return true if the resource name not exists , otherwise return false
* /
* /
public Result verifyResourceName ( String n ame, ResourceType type , User loginUser ) {
public Result verifyResourceName ( String fullN ame, ResourceType type , User loginUser ) {
Result result = new Result ( ) ;
Result result = new Result ( ) ;
putMsg ( result , Status . SUCCESS ) ;
putMsg ( result , Status . SUCCESS ) ;
if ( checkResourceExists ( n ame, 0 , type . ordinal ( ) ) ) {
if ( checkResourceExists ( fullN ame, 0 , type . ordinal ( ) ) ) {
logger . error ( "resource type:{} name:{} has exist, can't create again." , type , n ame) ;
logger . error ( "resource type:{} name:{} has exist, can't create again." , type , fullN ame) ;
putMsg ( result , Status . RESOURCE_EXIST ) ;
putMsg ( result , Status . RESOURCE_EXIST ) ;
} else {
} else {
// query tenant
// query tenant
@ -461,9 +662,9 @@ public class ResourcesService extends BaseService {
String tenantCode = tenant . getTenantCode ( ) ;
String tenantCode = tenant . getTenantCode ( ) ;
try {
try {
String hdfsFilename = getHdfsFileName ( type , tenantCode , n ame) ;
String hdfsFilename = HadoopUtils . getHdfsFileName ( type , tenantCode , fullN ame) ;
if ( HadoopUtils . getInstance ( ) . exists ( hdfsFilename ) ) {
if ( HadoopUtils . getInstance ( ) . exists ( hdfsFilename ) ) {
logger . error ( "resource type:{} name:{} has exist in hdfs {}, can't create again." , type , n ame, hdfsFilename ) ;
logger . error ( "resource type:{} name:{} has exist in hdfs {}, can't create again." , type , fullN ame, hdfsFilename ) ;
putMsg ( result , Status . RESOURCE_FILE_EXIST , hdfsFilename ) ;
putMsg ( result , Status . RESOURCE_FILE_EXIST , hdfsFilename ) ;
}
}
@ -480,6 +681,48 @@ public class ResourcesService extends BaseService {
return result ;
return result ;
}
}
/ * *
* verify resource by full name or pid and type
* @param fullName resource full name
* @param id resource id
* @param type resource type
* @return true if the resource full name or pid not exists , otherwise return false
* /
public Result queryResource ( String fullName , Integer id , ResourceType type ) {
Result result = new Result ( ) ;
if ( StringUtils . isBlank ( fullName ) & & id = = null ) {
logger . error ( "You must input one of fullName and pid" ) ;
putMsg ( result , Status . REQUEST_PARAMS_NOT_VALID_ERROR ) ;
return result ;
}
if ( StringUtils . isNotBlank ( fullName ) ) {
List < Resource > resourceList = resourcesMapper . queryResource ( fullName , type . ordinal ( ) ) ;
if ( CollectionUtils . isEmpty ( resourceList ) ) {
logger . error ( "resource file not exist, resource full name {} " , fullName ) ;
putMsg ( result , Status . RESOURCE_NOT_EXIST ) ;
return result ;
}
putMsg ( result , Status . SUCCESS ) ;
result . setData ( resourceList . get ( 0 ) ) ;
} else {
Resource resource = resourcesMapper . selectById ( id ) ;
if ( resource = = null ) {
logger . error ( "resource file not exist, resource id {}" , id ) ;
putMsg ( result , Status . RESOURCE_NOT_EXIST ) ;
return result ;
}
Resource parentResource = resourcesMapper . selectById ( resource . getPid ( ) ) ;
if ( parentResource = = null ) {
logger . error ( "parent resource file not exist, resource id {}" , id ) ;
putMsg ( result , Status . RESOURCE_NOT_EXIST ) ;
return result ;
}
putMsg ( result , Status . SUCCESS ) ;
result . setData ( parentResource ) ;
}
return result ;
}
/ * *
/ * *
* view resource file online
* view resource file online
*
*
@ -501,7 +744,7 @@ public class ResourcesService extends BaseService {
// get resource by id
// get resource by id
Resource resource = resourcesMapper . selectById ( resourceId ) ;
Resource resource = resourcesMapper . selectById ( resourceId ) ;
if ( resource = = null ) {
if ( resource = = null ) {
logger . error ( "resouce file not exist, resource id {}" , resourceId ) ;
logger . error ( "resour ce file not exist, resource id {}" , resourceId ) ;
putMsg ( result , Status . RESOURCE_NOT_EXIST ) ;
putMsg ( result , Status . RESOURCE_NOT_EXIST ) ;
return result ;
return result ;
}
}
@ -511,7 +754,7 @@ public class ResourcesService extends BaseService {
if ( StringUtils . isNotEmpty ( resourceViewSuffixs ) ) {
if ( StringUtils . isNotEmpty ( resourceViewSuffixs ) ) {
List < String > strList = Arrays . asList ( resourceViewSuffixs . split ( "," ) ) ;
List < String > strList = Arrays . asList ( resourceViewSuffixs . split ( "," ) ) ;
if ( ! strList . contains ( nameSuffix ) ) {
if ( ! strList . contains ( nameSuffix ) ) {
logger . error ( "resouce suffix {} not support view, resource id {}" , nameSuffix , resourceId ) ;
logger . error ( "resour ce suffix {} not support view, resource id {}" , nameSuffix , resourceId ) ;
putMsg ( result , Status . RESOURCE_SUFFIX_NOT_SUPPORT_VIEW ) ;
putMsg ( result , Status . RESOURCE_SUFFIX_NOT_SUPPORT_VIEW ) ;
return result ;
return result ;
}
}
@ -523,7 +766,7 @@ public class ResourcesService extends BaseService {
}
}
// hdfs path
// hdfs path
String hdfsFileName = HadoopUtils . getHdfsFilen ame ( tenantCode , resource . getAlias ( ) ) ;
String hdfsFileName = HadoopUtils . getHdfsResourceFileN ame ( tenantCode , resource . getFullName ( ) ) ;
logger . info ( "resource hdfs path is {} " , hdfsFileName ) ;
logger . info ( "resource hdfs path is {} " , hdfsFileName ) ;
try {
try {
if ( HadoopUtils . getInstance ( ) . exists ( hdfsFileName ) ) {
if ( HadoopUtils . getInstance ( ) . exists ( hdfsFileName ) ) {
@ -559,7 +802,7 @@ public class ResourcesService extends BaseService {
* @return create result code
* @return create result code
* /
* /
@Transactional ( rollbackFor = Exception . class )
@Transactional ( rollbackFor = Exception . class )
public Result onlineCreateResource ( User loginUser , ResourceType type , String fileName , String fileSuffix , String desc , String content ) {
public Result onlineCreateResource ( User loginUser , ResourceType type , String fileName , String fileSuffix , String desc , String content , int pid , String currentDirectory ) {
Result result = new Result ( ) ;
Result result = new Result ( ) ;
// if resource upload startup
// if resource upload startup
if ( ! PropertyUtils . getResUploadStartupState ( ) ) {
if ( ! PropertyUtils . getResUploadStartupState ( ) ) {
@ -581,15 +824,16 @@ public class ResourcesService extends BaseService {
}
}
String name = fileName . trim ( ) + "." + nameSuffix ;
String name = fileName . trim ( ) + "." + nameSuffix ;
String fullName = currentDirectory . equals ( "/" ) ? String . format ( "%s%s" , currentDirectory , name ) : String . format ( "%s/%s" , currentDirectory , name ) ;
result = verifyResourceName ( n ame, type , loginUser ) ;
result = verifyResourceName ( fullN ame, type , loginUser ) ;
if ( ! result . getCode ( ) . equals ( Status . SUCCESS . getCode ( ) ) ) {
if ( ! result . getCode ( ) . equals ( Status . SUCCESS . getCode ( ) ) ) {
return result ;
return result ;
}
}
// save data
// save data
Date now = new Date ( ) ;
Date now = new Date ( ) ;
Resource resource = new Resource ( name , name , desc , loginUser . getId ( ) , type , content . getBytes ( ) . length , now , now ) ;
Resource resource = new Resource ( pid , name , fullName , false , desc , name , loginUser . getId ( ) , type , content . getBytes ( ) . length , now , now ) ;
resourcesMapper . insert ( resource ) ;
resourcesMapper . insert ( resource ) ;
@ -605,7 +849,7 @@ public class ResourcesService extends BaseService {
String tenantCode = tenantMapper . queryById ( loginUser . getTenantId ( ) ) . getTenantCode ( ) ;
String tenantCode = tenantMapper . queryById ( loginUser . getTenantId ( ) ) . getTenantCode ( ) ;
result = uploadContentToHdfs ( n ame, tenantCode , content ) ;
result = uploadContentToHdfs ( fullN ame, tenantCode , content ) ;
if ( ! result . getCode ( ) . equals ( Status . SUCCESS . getCode ( ) ) ) {
if ( ! result . getCode ( ) . equals ( Status . SUCCESS . getCode ( ) ) ) {
throw new RuntimeException ( result . getMsg ( ) ) ;
throw new RuntimeException ( result . getMsg ( ) ) ;
}
}
@ -657,7 +901,7 @@ public class ResourcesService extends BaseService {
resourcesMapper . updateById ( resource ) ;
resourcesMapper . updateById ( resource ) ;
result = uploadContentToHdfs ( resource . getAlias ( ) , tenantCode , content ) ;
result = uploadContentToHdfs ( resource . getFullName ( ) , tenantCode , content ) ;
if ( ! result . getCode ( ) . equals ( Status . SUCCESS . getCode ( ) ) ) {
if ( ! result . getCode ( ) . equals ( Status . SUCCESS . getCode ( ) ) ) {
throw new RuntimeException ( result . getMsg ( ) ) ;
throw new RuntimeException ( result . getMsg ( ) ) ;
}
}
@ -665,10 +909,10 @@ public class ResourcesService extends BaseService {
}
}
/ * *
/ * *
* @param resourceName
* @param resourceName resource name
* @param tenantCode
* @param tenantCode tenant code
* @param content
* @param content content
* @return
* @return result
* /
* /
private Result uploadContentToHdfs ( String resourceName , String tenantCode , String content ) {
private Result uploadContentToHdfs ( String resourceName , String tenantCode , String content ) {
Result result = new Result ( ) ;
Result result = new Result ( ) ;
@ -684,8 +928,8 @@ public class ResourcesService extends BaseService {
return result ;
return result ;
}
}
// get file hdfs path
// get resource file hdfs path
hdfsFileName = HadoopUtils . getHdfsFilen ame ( tenantCode , resourceName ) ;
hdfsFileName = HadoopUtils . getHdfsResourceFileN ame ( tenantCode , resourceName ) ;
String resourcePath = HadoopUtils . getHdfsResDir ( tenantCode ) ;
String resourcePath = HadoopUtils . getHdfsResDir ( tenantCode ) ;
logger . info ( "resource hdfs path is {} " , hdfsFileName ) ;
logger . info ( "resource hdfs path is {} " , hdfsFileName ) ;
@ -729,11 +973,14 @@ public class ResourcesService extends BaseService {
logger . error ( "download file not exist, resource id {}" , resourceId ) ;
logger . error ( "download file not exist, resource id {}" , resourceId ) ;
return null ;
return null ;
}
}
if ( resource . isDirectory ( ) ) {
logger . error ( "resource id {} is directory,can't download it" , resourceId ) ;
throw new RuntimeException ( "cant't download directory" ) ;
}
User user = userMapper . queryDetailsById ( resource . getUserId ( ) ) ;
User user = userMapper . queryDetailsById ( resource . getUserId ( ) ) ;
String tenantCode = tenantMapper . queryById ( user . getTenantId ( ) ) . getTenantCode ( ) ;
String tenantCode = tenantMapper . queryById ( user . getTenantId ( ) ) . getTenantCode ( ) ;
String hdfsFileName = "" ;
String hdfsFileName = HadoopUtils . getHdfsFileName ( resource . getType ( ) , tenantCode , resource . getAlias ( ) ) ;
hdfsFileName = getHdfsFileName ( resource , tenantCode , hdfsFileName ) ;
String localFileName = FileUtils . getDownloadFilename ( resource . getAlias ( ) ) ;
String localFileName = FileUtils . getDownloadFilename ( resource . getAlias ( ) ) ;
logger . info ( "resource hdfs path is {} " , hdfsFileName ) ;
logger . info ( "resource hdfs path is {} " , hdfsFileName ) ;
@ -743,6 +990,33 @@ public class ResourcesService extends BaseService {
}
}
/ * *
* list all file
*
* @param loginUser login user
* @param userId user id
* @return unauthorized result code
* /
public Map < String , Object > authorizeResourceTree ( User loginUser , Integer userId ) {
Map < String , Object > result = new HashMap < > ( ) ;
if ( checkAdmin ( loginUser , result ) ) {
return result ;
}
List < Resource > resourceList = resourcesMapper . queryResourceExceptUserId ( userId ) ;
List < ResourceComponent > list ;
if ( CollectionUtils . isNotEmpty ( resourceList ) ) {
Visitor visitor = new ResourceTreeVisitor ( resourceList ) ;
list = visitor . visit ( ) . getChildren ( ) ;
} else {
list = new ArrayList < > ( 0 ) ;
}
result . put ( Constants . DATA_LIST , list ) ;
putMsg ( result , Status . SUCCESS ) ;
return result ;
}
/ * *
/ * *
* unauthorized file
* unauthorized file
*
*
@ -757,8 +1031,8 @@ public class ResourcesService extends BaseService {
return result ;
return result ;
}
}
List < Resource > resourceList = resourcesMapper . queryResourceExceptUserId ( userId ) ;
List < Resource > resourceList = resourcesMapper . queryResourceExceptUserId ( userId ) ;
List < Object > list ;
List < Resource > list ;
if ( CollectionUtils . isNotEmpty ( resourceList ) ) {
if ( resourceList ! = null & & resourceList . size ( ) > 0 ) {
Set < Resource > resourceSet = new HashSet < > ( resourceList ) ;
Set < Resource > resourceSet = new HashSet < > ( resourceList ) ;
List < Resource > authedResourceList = resourcesMapper . queryAuthorizedResourceList ( userId ) ;
List < Resource > authedResourceList = resourcesMapper . queryAuthorizedResourceList ( userId ) ;
@ -767,15 +1041,12 @@ public class ResourcesService extends BaseService {
} else {
} else {
list = new ArrayList < > ( 0 ) ;
list = new ArrayList < > ( 0 ) ;
}
}
Visitor visitor = new ResourceTreeVisitor ( list ) ;
result . put ( Constants . DATA_LIST , l ist) ;
result . put ( Constants . DATA_LIST , visitor . v isi t( ) . getChildren ( ) ) ;
putMsg ( result , Status . SUCCESS ) ;
putMsg ( result , Status . SUCCESS ) ;
return result ;
return result ;
}
}
/ * *
/ * *
* unauthorized udf function
* unauthorized udf function
*
*
@ -841,46 +1112,15 @@ public class ResourcesService extends BaseService {
return result ;
return result ;
}
}
List < Resource > authedResources = resourcesMapper . queryAuthorizedResourceList ( userId ) ;
List < Resource > authedResources = resourcesMapper . queryAuthorizedResourceList ( userId ) ;
Visitor visitor = new ResourceTreeVisitor ( authedResources ) ;
result . put ( Constants . DATA_LIST , authedResources ) ;
logger . info ( JSON . toJSONString ( visitor . visit ( ) , SerializerFeature . SortField ) ) ;
String jsonTreeStr = JSON . toJSONString ( visitor . visit ( ) . getChildren ( ) , SerializerFeature . SortField ) ;
logger . info ( jsonTreeStr ) ;
result . put ( Constants . DATA_LIST , visitor . visit ( ) . getChildren ( ) ) ;
putMsg ( result , Status . SUCCESS ) ;
putMsg ( result , Status . SUCCESS ) ;
return result ;
return result ;
}
}
/ * *
* get hdfs file name
*
* @param resource resource
* @param tenantCode tenant code
* @param hdfsFileName hdfs file name
* @return hdfs file name
* /
private String getHdfsFileName ( Resource resource , String tenantCode , String hdfsFileName ) {
if ( resource . getType ( ) . equals ( ResourceType . FILE ) ) {
hdfsFileName = HadoopUtils . getHdfsFilename ( tenantCode , resource . getAlias ( ) ) ;
} else if ( resource . getType ( ) . equals ( ResourceType . UDF ) ) {
hdfsFileName = HadoopUtils . getHdfsUdfFilename ( tenantCode , resource . getAlias ( ) ) ;
}
return hdfsFileName ;
}
/ * *
* get hdfs file name
*
* @param resourceType resource type
* @param tenantCode tenant code
* @param hdfsFileName hdfs file name
* @return hdfs file name
* /
private String getHdfsFileName ( ResourceType resourceType , String tenantCode , String hdfsFileName ) {
if ( resourceType . equals ( ResourceType . FILE ) ) {
hdfsFileName = HadoopUtils . getHdfsFilename ( tenantCode , hdfsFileName ) ;
} else if ( resourceType . equals ( ResourceType . UDF ) ) {
hdfsFileName = HadoopUtils . getHdfsUdfFilename ( tenantCode , hdfsFileName ) ;
}
return hdfsFileName ;
}
/ * *
/ * *
* get authorized resource list
* get authorized resource list
*
*
@ -920,4 +1160,69 @@ public class ResourcesService extends BaseService {
return tenant . getTenantCode ( ) ;
return tenant . getTenantCode ( ) ;
}
}
/ * *
* list all children id
* @param resource resource
* @return all children id
* /
List < Integer > listAllChildren ( Resource resource ) {
List < Integer > childList = new ArrayList < > ( ) ;
if ( resource . getId ( ) ! = - 1 ) {
childList . add ( resource . getId ( ) ) ;
}
if ( resource . isDirectory ( ) ) {
listAllChildren ( resource . getId ( ) , childList ) ;
}
return childList ;
}
/ * *
* list all children id
* @param resourceId resource id
* @param childList child list
* /
void listAllChildren ( int resourceId , List < Integer > childList ) {
List < Integer > children = resourcesMapper . listChildren ( resourceId ) ;
for ( int chlidId : children ) {
childList . add ( chlidId ) ;
listAllChildren ( chlidId , childList ) ;
}
}
/ * *
* get resource process map key is resource id , value is the set of process definition
* @return resource process definition map
* /
private Map < Integer , Set < Integer > > getResourceProcessMap ( ) {
Map < Integer , String > map = new HashMap < > ( ) ;
Map < Integer , Set < Integer > > result = new HashMap < > ( ) ;
List < Map < String , Object > > list = processDefinitionMapper . listResources ( ) ;
if ( CollectionUtils . isNotEmpty ( list ) ) {
for ( Map < String , Object > tempMap : list ) {
map . put ( ( Integer ) tempMap . get ( "id" ) , ( String ) tempMap . get ( "resource_ids" ) ) ;
}
}
for ( Map . Entry < Integer , String > entry : map . entrySet ( ) ) {
Integer mapKey = entry . getKey ( ) ;
String [ ] arr = entry . getValue ( ) . split ( "," ) ;
Set < Integer > mapValues = Arrays . stream ( arr ) . map ( Integer : : parseInt ) . collect ( Collectors . toSet ( ) ) ;
for ( Integer value : mapValues ) {
if ( result . containsKey ( value ) ) {
Set < Integer > set = result . get ( value ) ;
set . add ( mapKey ) ;
result . put ( value , set ) ;
} else {
Set < Integer > set = new HashSet < > ( ) ;
set . add ( mapKey ) ;
result . put ( value , set ) ;
}
}
}
return result ;
}
}
}