[improv][Resource Center] Implement getResourceFileName in StorageOperator (#14097)

dolphinscheduler-api/src/main/java/org/apache/dolphinscheduler/api/service/impl/ResourcesServiceImpl.java

@@ -1534,7 +1534,7 @@ public class ResourcesServiceImpl extends BaseServiceImpl implements ResourcesSe
 
         String tenantCode = getTenantCode(user);
 
-        String baseFolder = storageOperate.getResourceFileName(tenantCode, "DATA_TRANSFER");
+        String baseFolder = storageOperate.getResourceFullName(tenantCode, "DATA_TRANSFER");
 
         LocalDateTime now = LocalDateTime.now();
         now = now.minus(days, ChronoUnit.DAYS);

dolphinscheduler-dao/src/main/java/org/apache/dolphinscheduler/dao/utils/TaskCacheUtils.java

@@ -178,7 +178,7 @@ public class TaskCacheUtils {
     public static String getValCheckSum(Property fileProperty, TaskExecutionContext context,
                                         StorageOperate storageOperate) {
         String resourceCRCPath = fileProperty.getValue() + CRC_SUFFIX;
-        String resourceCRCWholePath = storageOperate.getResourceFileName(context.getTenantCode(), resourceCRCPath);
+        String resourceCRCWholePath = storageOperate.getResourceFullName(context.getTenantCode(), resourceCRCPath);
         String targetPath = String.format("%s/%s", context.getExecutePath(), resourceCRCPath);
         log.info("{} --- Remote:{} to Local:{}", "CRC file", resourceCRCWholePath, targetPath);
         String crcString = "";

dolphinscheduler-storage-plugin/dolphinscheduler-storage-api/src/main/java/org/apache/dolphinscheduler/plugin/storage/api/StorageOperate.java

@@ -60,19 +60,19 @@ public interface StorageOperate {
     boolean mkdir(String tenantCode, String path) throws IOException;
 
     /**
-     * get the path of the resource file
+     * get the path of the resource file (fullName)
      * @param tenantCode
-     * @param fullName
+     * @param fileName
      * @return
      */
-    String getResourceFileName(String tenantCode, String fullName);
+    String getResourceFullName(String tenantCode, String fileName);
 
     /**
-     * get the path of the resource file excluding the base path.
+     * get the path of the resource file excluding the base path (fileName)
      * @param fullName
      * @return
      */
-    String getResourceFileName(String fullName);
+    String getResourceFileName(String tenantCode, String fullName);
 
     /**
      * get the path of the file

dolphinscheduler-storage-plugin/dolphinscheduler-storage-gcs/src/main/java/org/apache/dolphinscheduler/plugin/storage/gcs/GcsStorageOperator.java

@@ -122,7 +122,7 @@ public class GcsStorageOperator implements Closeable, StorageOperate {
     }
 
     @Override
-    public String getResourceFileName(String tenantCode, String fileName) {
+    public String getResourceFullName(String tenantCode, String fileName) {
         if (fileName.startsWith(FOLDER_SEPARATOR)) {
             fileName.replaceFirst(FOLDER_SEPARATOR, EMPTY_STRING);
         }
@@ -130,8 +130,9 @@ public class GcsStorageOperator implements Closeable, StorageOperate {
     }
 
     @Override
-    public String getResourceFileName(String fullName) {
+    public String getResourceFileName(String tenantCode, String fullName) {
-        return null;
+        String resDir = getResDir(tenantCode);
+        return fullName.replaceFirst(resDir, "");
     }
 
     @Override

dolphinscheduler-storage-plugin/dolphinscheduler-storage-gcs/src/test/java/org/apache/dolphinscheduler/plugin/storage/gcs/GcsStorageOperatorTest.java

@@ -146,11 +146,20 @@ public class GcsStorageOperatorTest {
         Assertions.assertTrue(isSuccess);
     }
 
+    @Test
+    public void getResourceFullName() {
+        final String expectedResourceFullName =
+                String.format("dolphinscheduler/%s/resources/%s", TENANT_CODE_MOCK, FILE_NAME_MOCK);
+        final String resourceFullName = gcsStorageOperator.getResourceFullName(TENANT_CODE_MOCK, FILE_NAME_MOCK);
+        Assertions.assertEquals(expectedResourceFullName, resourceFullName);
+    }
+
     @Test
     public void getResourceFileName() {
-        final String expectedResourceFileName =
+        final String expectedResourceFileName = FILE_NAME_MOCK;
+        final String resourceFullName =
                 String.format("dolphinscheduler/%s/resources/%s", TENANT_CODE_MOCK, FILE_NAME_MOCK);
-        final String resourceFileName = gcsStorageOperator.getResourceFileName(TENANT_CODE_MOCK, FILE_NAME_MOCK);
+        final String resourceFileName = gcsStorageOperator.getResourceFileName(TENANT_CODE_MOCK, resourceFullName);
         Assertions.assertEquals(expectedResourceFileName, resourceFileName);
     }
 

dolphinscheduler-storage-plugin/dolphinscheduler-storage-hdfs/src/main/java/org/apache/dolphinscheduler/plugin/storage/hdfs/HdfsStorageOperator.java

@@ -56,7 +56,6 @@ import java.nio.charset.StandardCharsets;
 import java.nio.file.Files;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.Date;
 import java.util.LinkedList;
@@ -69,7 +68,6 @@ import java.util.stream.Stream;
 import lombok.extern.slf4j.Slf4j;
 
 import com.fasterxml.jackson.databind.node.ObjectNode;
-import com.google.common.base.Joiner;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheLoader;
 import com.google.common.cache.LoadingCache;
@@ -298,28 +296,14 @@ public class HdfsStorageOperator implements Closeable, StorageOperate {
     }
 
     @Override
-    public String getResourceFileName(String tenantCode, String fullName) {
+    public String getResourceFullName(String tenantCode, String fullName) {
         return getHdfsResourceFileName(tenantCode, fullName);
     }
 
     @Override
-    public String getResourceFileName(String fullName) {
+    public String getResourceFileName(String tenantCode, String fullName) {
-        // here is a quick fix here to get fileName. We get the resource upload path and
+        String resDir = getResDir(tenantCode);
-        // get the index of the first appearance of resource upload path. The index is put
+        return fullName.replaceFirst(resDir, "");
-        // in the start index of the substring function and get the result substring containing
-        // tenantcode and "resource" directory and the fileName.
-        // Then we split the result substring
-        // with "/" and join all elements except the first two elements because they are
-        // tenantCode and "resource" directory.
-        String resourceUploadPath =
-                RESOURCE_UPLOAD_PATH.endsWith(FOLDER_SEPARATOR) ? StringUtils.chop(RESOURCE_UPLOAD_PATH)
-                        : RESOURCE_UPLOAD_PATH;
-        // +1 because we want to skip the "/" after resource upload path as well.
-        String pathContainingTenantNResource = fullName.substring(
-                fullName.indexOf(resourceUploadPath)
-                        + resourceUploadPath.length() + 1);
-        String[] fileNameArr = pathContainingTenantNResource.split(FOLDER_SEPARATOR);
-        return Joiner.on(FOLDER_SEPARATOR).join(Arrays.stream(fileNameArr).skip(2).collect(Collectors.toList()));
     }
 
     @Override

dolphinscheduler-storage-plugin/dolphinscheduler-storage-oss/src/main/java/org/apache/dolphinscheduler/plugin/storage/oss/OssStorageOperator.java

@@ -175,7 +175,7 @@ public class OssStorageOperator implements Closeable, StorageOperate {
     }
 
     @Override
-    public String getResourceFileName(String tenantCode, String fileName) {
+    public String getResourceFullName(String tenantCode, String fileName) {
         if (fileName.startsWith(FOLDER_SEPARATOR)) {
             fileName = fileName.replaceFirst(FOLDER_SEPARATOR, "");
         }
@@ -183,8 +183,9 @@ public class OssStorageOperator implements Closeable, StorageOperate {
     }
 
     @Override
-    public String getResourceFileName(String fullName) {
+    public String getResourceFileName(String tenantCode, String fullName) {
-        return null;
+        String resDir = getResDir(tenantCode);
+        return fullName.replaceFirst(resDir, "");
     }
 
     @Override

dolphinscheduler-storage-plugin/dolphinscheduler-storage-oss/src/test/java/org/apache/dolphinscheduler/plugin/storage/oss/OssStorageOperatorTest.java

@@ -158,11 +158,20 @@ public class OssStorageOperatorTest {
         Assertions.assertTrue(isSuccess);
     }
 
+    @Test
+    public void getResourceFullName() {
+        final String expectedResourceFullName =
+                String.format("dolphinscheduler/%s/resources/%s", TENANT_CODE_MOCK, FILE_NAME_MOCK);
+        final String resourceFullName = ossOperator.getResourceFullName(TENANT_CODE_MOCK, FILE_NAME_MOCK);
+        Assertions.assertEquals(expectedResourceFullName, resourceFullName);
+    }
+
     @Test
     public void getResourceFileName() {
-        final String expectedResourceFileName =
+        final String expectedResourceFileName = FILE_NAME_MOCK;
+        final String resourceFullName =
                 String.format("dolphinscheduler/%s/resources/%s", TENANT_CODE_MOCK, FILE_NAME_MOCK);
-        final String resourceFileName = ossOperator.getResourceFileName(TENANT_CODE_MOCK, FILE_NAME_MOCK);
+        final String resourceFileName = ossOperator.getResourceFileName(TENANT_CODE_MOCK, resourceFullName);
         Assertions.assertEquals(expectedResourceFileName, resourceFileName);
     }
 

dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/main/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperator.java

@@ -46,7 +46,6 @@ import java.nio.file.Files;
 import java.nio.file.Path;
 import java.nio.file.Paths;
 import java.util.ArrayList;
-import java.util.Arrays;
 import java.util.Collections;
 import java.util.LinkedList;
 import java.util.List;
@@ -76,7 +75,6 @@ import com.amazonaws.services.s3.model.S3ObjectSummary;
 import com.amazonaws.services.s3.transfer.MultipleFileDownload;
 import com.amazonaws.services.s3.transfer.TransferManager;
 import com.amazonaws.services.s3.transfer.TransferManagerBuilder;
-import com.google.common.base.Joiner;
 
 @Slf4j
 @Data
@@ -182,7 +180,7 @@ public class S3StorageOperator implements Closeable, StorageOperate {
     }
 
     @Override
-    public String getResourceFileName(String tenantCode, String fileName) {
+    public String getResourceFullName(String tenantCode, String fileName) {
         if (fileName.startsWith(FOLDER_SEPARATOR)) {
             fileName = fileName.replaceFirst(FOLDER_SEPARATOR, "");
         }
@@ -190,23 +188,9 @@ public class S3StorageOperator implements Closeable, StorageOperate {
     }
 
     @Override
-    public String getResourceFileName(String fullName) {
+    public String getResourceFileName(String tenantCode, String fullName) {
-        // here is a quick fix here to get fileName. We get the resource upload path and
+        String resDir = getResDir(tenantCode);
-        // get the index of the first appearance of resource upload path. The index is put
+        return fullName.replaceFirst(resDir, "");
-        // in the start index of the substring function and get the result substring containing
-        // tenantcode and "resource" directory and the fileName.
-        // Then we split the result substring
-        // with "/" and join all elements except the first two elements because they are
-        // tenantCode and "resource" directory.
-        String resourceUploadPath =
-                RESOURCE_UPLOAD_PATH.endsWith(FOLDER_SEPARATOR) ? StringUtils.chop(RESOURCE_UPLOAD_PATH)
-                        : RESOURCE_UPLOAD_PATH;
-        // +1 because we want to skip the "/" after resource upload path as well.
-        String pathContainingTenantNResource = fullName.substring(
-                fullName.indexOf(resourceUploadPath)
-                        + resourceUploadPath.length() + 1);
-        String[] fileNameArr = pathContainingTenantNResource.split(FOLDER_SEPARATOR);
-        return Joiner.on(FOLDER_SEPARATOR).join(Arrays.stream(fileNameArr).skip(2).collect(Collectors.toList()));
     }
 
     @Override

dolphinscheduler-storage-plugin/dolphinscheduler-storage-s3/src/test/java/org/apache/dolphinscheduler/plugin/storage/s3/S3StorageOperatorTest.java

@@ -164,11 +164,20 @@ public class S3StorageOperatorTest {
         Assertions.assertTrue(isSuccess);
     }
 
+    @Test
+    public void getResourceFullName() {
+        final String expectedResourceFullName =
+                String.format("dolphinscheduler/%s/resources/%s", TENANT_CODE_MOCK, FILE_NAME_MOCK);
+        final String resourceFullName = s3StorageOperator.getResourceFullName(TENANT_CODE_MOCK, FILE_NAME_MOCK);
+        Assertions.assertEquals(expectedResourceFullName, resourceFullName);
+    }
+
     @Test
     public void getResourceFileName() {
-        final String expectedResourceFileName =
+        final String expectedResourceFileName = FILE_NAME_MOCK;
+        final String resourceFullName =
                 String.format("dolphinscheduler/%s/resources/%s", TENANT_CODE_MOCK, FILE_NAME_MOCK);
-        final String resourceFileName = s3StorageOperator.getResourceFileName(TENANT_CODE_MOCK, FILE_NAME_MOCK);
+        final String resourceFileName = s3StorageOperator.getResourceFileName(TENANT_CODE_MOCK, resourceFullName);
         Assertions.assertEquals(expectedResourceFileName, resourceFileName);
     }
 

dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskExecutionCheckerUtils.java

@@ -109,16 +109,19 @@ public class TaskExecutionCheckerUtils {
     public static void downloadResourcesIfNeeded(StorageOperate storageOperate,
                                                  TaskExecutionContext taskExecutionContext) {
         String execLocalPath = taskExecutionContext.getExecutePath();
+        String tenant = taskExecutionContext.getTenantCode();
         Map<String, String> projectRes = taskExecutionContext.getResources();
         if (MapUtils.isEmpty(projectRes)) {
             return;
         }
         List<Pair<String, String>> downloadFiles = new ArrayList<>();
-        projectRes.forEach((key, value) -> {
+        projectRes.keySet().forEach(fullName -> {
-            File resFile = new File(execLocalPath, key);
+            String fileName = storageOperate.getResourceFileName(tenant, fullName);
+            projectRes.put(fullName, fileName);
+            File resFile = new File(execLocalPath, fileName);
             boolean notExist = !resFile.exists();
             if (notExist) {
-                downloadFiles.add(Pair.of(key, value));
+                downloadFiles.add(Pair.of(fullName, fileName));
             } else {
                 log.info("file : {} exists ", resFile.getName());
             }
@@ -131,10 +134,7 @@ public class TaskExecutionCheckerUtils {
             for (Pair<String, String> fileDownload : downloadFiles) {
                 try {
                     String fullName = fileDownload.getLeft();
-                    // we do not actually get & need tenantCode with this implementation right now.
+                    String fileName = fileDownload.getRight();
-                    String tenantCode = fileDownload.getRight();
-                    // TODO: Need a better way to get fileName because this implementation is tricky.
-                    String fileName = storageOperate.getResourceFileName(fullName);
                     log.info("get resource file from path:{}", fullName);
 
                     long resourceDownloadStartTime = System.currentTimeMillis();

dolphinscheduler-worker/src/main/java/org/apache/dolphinscheduler/server/worker/utils/TaskFilesTransferUtils.java

@@ -101,9 +101,9 @@ public class TaskFilesTransferUtils {
             try {
                 // upload file to storage
                 String resourceWholePath =
-                        storageOperate.getResourceFileName(taskExecutionContext.getTenantCode(), resourcePath);
+                        storageOperate.getResourceFullName(taskExecutionContext.getTenantCode(), resourcePath);
                 String resourceCRCWholePath =
-                        storageOperate.getResourceFileName(taskExecutionContext.getTenantCode(), resourceCRCPath);
+                        storageOperate.getResourceFullName(taskExecutionContext.getTenantCode(), resourceCRCPath);
                 log.info("{} --- Local:{} to Remote:{}", property, srcPath, resourceWholePath);
                 storageOperate.upload(taskExecutionContext.getTenantCode(), srcPath, resourceWholePath, false, true);
                 log.info("{} --- Local:{} to Remote:{}", "CRC file", srcCRCPath, resourceCRCWholePath);
@@ -176,7 +176,7 @@ public class TaskFilesTransferUtils {
 
             try {
                 String resourceWholePath =
-                        storageOperate.getResourceFileName(taskExecutionContext.getTenantCode(), resourcePath);
+                        storageOperate.getResourceFullName(taskExecutionContext.getTenantCode(), resourcePath);
                 log.info("{} --- Remote:{} to Local:{}", property, resourceWholePath, downloadPath);
                 storageOperate.download(taskExecutionContext.getTenantCode(), resourceWholePath, downloadPath, false,
                         true);