|
|
@@ -17,28 +17,21 @@
|
|
|
|
|
|
package org.apache.dolphinscheduler.common.utils;
|
|
|
|
|
|
-import com.amazonaws.AmazonServiceException;
|
|
|
-import com.amazonaws.auth.AWSStaticCredentialsProvider;
|
|
|
-import com.amazonaws.auth.BasicAWSCredentials;
|
|
|
-import com.amazonaws.client.builder.AwsClientBuilder;
|
|
|
-import com.amazonaws.regions.Regions;
|
|
|
-import com.amazonaws.services.s3.AmazonS3;
|
|
|
-import com.amazonaws.services.s3.AmazonS3ClientBuilder;
|
|
|
-import com.amazonaws.services.s3.model.AmazonS3Exception;
|
|
|
-import com.amazonaws.services.s3.model.ObjectMetadata;
|
|
|
-import com.amazonaws.services.s3.model.PutObjectRequest;
|
|
|
-import com.amazonaws.services.s3.model.S3Object;
|
|
|
-import com.amazonaws.services.s3.model.S3ObjectInputStream;
|
|
|
-import com.amazonaws.services.s3.transfer.MultipleFileDownload;
|
|
|
-import com.amazonaws.services.s3.transfer.TransferManager;
|
|
|
-import com.amazonaws.services.s3.transfer.TransferManagerBuilder;
|
|
|
-import org.apache.commons.lang3.StringUtils;
|
|
|
+import static org.apache.dolphinscheduler.common.Constants.AWS_END_POINT;
|
|
|
+import static org.apache.dolphinscheduler.common.Constants.FOLDER_SEPARATOR;
|
|
|
+import static org.apache.dolphinscheduler.common.Constants.FORMAT_S_S;
|
|
|
+import static org.apache.dolphinscheduler.common.Constants.RESOURCE_STORAGE_TYPE;
|
|
|
+import static org.apache.dolphinscheduler.common.Constants.RESOURCE_TYPE_FILE;
|
|
|
+import static org.apache.dolphinscheduler.common.Constants.RESOURCE_TYPE_UDF;
|
|
|
+import static org.apache.dolphinscheduler.common.Constants.STORAGE_S3;
|
|
|
+
|
|
|
+import org.apache.dolphinscheduler.common.Constants;
|
|
|
import org.apache.dolphinscheduler.common.enums.ResUploadType;
|
|
|
import org.apache.dolphinscheduler.common.storage.StorageOperate;
|
|
|
import org.apache.dolphinscheduler.plugin.task.api.TaskConstants;
|
|
|
import org.apache.dolphinscheduler.spi.enums.ResourceType;
|
|
|
-import org.slf4j.Logger;
|
|
|
-import org.slf4j.LoggerFactory;
|
|
|
+
|
|
|
+import org.apache.commons.lang3.StringUtils;
|
|
|
|
|
|
import java.io.BufferedReader;
|
|
|
import java.io.ByteArrayInputStream;
|
|
|
@@ -54,14 +47,25 @@ import java.util.List;
|
|
|
import java.util.stream.Collectors;
|
|
|
import java.util.stream.Stream;
|
|
|
|
|
|
-import static org.apache.dolphinscheduler.common.Constants.AWS_END_POINT;
|
|
|
-import static org.apache.dolphinscheduler.common.Constants.BUCKET_NAME;
|
|
|
-import static org.apache.dolphinscheduler.common.Constants.FOLDER_SEPARATOR;
|
|
|
-import static org.apache.dolphinscheduler.common.Constants.FORMAT_S_S;
|
|
|
-import static org.apache.dolphinscheduler.common.Constants.RESOURCE_STORAGE_TYPE;
|
|
|
-import static org.apache.dolphinscheduler.common.Constants.RESOURCE_TYPE_FILE;
|
|
|
-import static org.apache.dolphinscheduler.common.Constants.RESOURCE_TYPE_UDF;
|
|
|
-import static org.apache.dolphinscheduler.common.Constants.STORAGE_S3;
|
|
|
+import org.slf4j.Logger;
|
|
|
+import org.slf4j.LoggerFactory;
|
|
|
+
|
|
|
+import com.amazonaws.AmazonServiceException;
|
|
|
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
|
|
|
+import com.amazonaws.auth.BasicAWSCredentials;
|
|
|
+import com.amazonaws.client.builder.AwsClientBuilder;
|
|
|
+import com.amazonaws.regions.Regions;
|
|
|
+import com.amazonaws.services.s3.AmazonS3;
|
|
|
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
|
|
|
+import com.amazonaws.services.s3.model.AmazonS3Exception;
|
|
|
+import com.amazonaws.services.s3.model.Bucket;
|
|
|
+import com.amazonaws.services.s3.model.ObjectMetadata;
|
|
|
+import com.amazonaws.services.s3.model.PutObjectRequest;
|
|
|
+import com.amazonaws.services.s3.model.S3Object;
|
|
|
+import com.amazonaws.services.s3.model.S3ObjectInputStream;
|
|
|
+import com.amazonaws.services.s3.transfer.MultipleFileDownload;
|
|
|
+import com.amazonaws.services.s3.transfer.TransferManager;
|
|
|
+import com.amazonaws.services.s3.transfer.TransferManagerBuilder;
|
|
|
|
|
|
public class S3Utils implements Closeable, StorageOperate {
|
|
|
|
|
|
@@ -73,6 +77,8 @@ public class S3Utils implements Closeable, StorageOperate {
|
|
|
|
|
|
public static final String REGION = PropertyUtils.getString(TaskConstants.AWS_REGION);
|
|
|
|
|
|
+ public static final String BUCKET_NAME = PropertyUtils.getString(Constants.AWS_S3_BUCKET_NAME);
|
|
|
+
|
|
|
|
|
|
private AmazonS3 s3Client = null;
|
|
|
|
|
|
@@ -81,19 +87,19 @@ public class S3Utils implements Closeable, StorageOperate {
|
|
|
|
|
|
if (!StringUtils.isEmpty(PropertyUtils.getString(AWS_END_POINT))) {
|
|
|
s3Client = AmazonS3ClientBuilder
|
|
|
- .standard()
|
|
|
- .withPathStyleAccessEnabled(true)
|
|
|
- .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(PropertyUtils.getString(AWS_END_POINT), Regions.fromName(REGION).getName()))
|
|
|
- .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(ACCESS_KEY_ID, SECRET_KEY_ID)))
|
|
|
- .build();
|
|
|
+ .standard()
|
|
|
+ .withPathStyleAccessEnabled(true)
|
|
|
+ .withEndpointConfiguration(new AwsClientBuilder.EndpointConfiguration(PropertyUtils.getString(AWS_END_POINT), Regions.fromName(REGION).getName()))
|
|
|
+ .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(ACCESS_KEY_ID, SECRET_KEY_ID)))
|
|
|
+ .build();
|
|
|
} else {
|
|
|
s3Client = AmazonS3ClientBuilder
|
|
|
- .standard()
|
|
|
- .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(ACCESS_KEY_ID, SECRET_KEY_ID)))
|
|
|
- .withRegion(Regions.fromName(REGION))
|
|
|
- .build();
|
|
|
+ .standard()
|
|
|
+ .withCredentials(new AWSStaticCredentialsProvider(new BasicAWSCredentials(ACCESS_KEY_ID, SECRET_KEY_ID)))
|
|
|
+ .withRegion(Regions.fromName(REGION))
|
|
|
+ .build();
|
|
|
}
|
|
|
- checkBucketNameIfNotPresent(BUCKET_NAME);
|
|
|
+ checkBucketNameExists(BUCKET_NAME);
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -125,24 +131,31 @@ public class S3Utils implements Closeable, StorageOperate {
|
|
|
|
|
|
@Override
|
|
|
public void createTenantDirIfNotExists(String tenantCode) throws Exception {
|
|
|
- createFolder(tenantCode+ FOLDER_SEPARATOR +RESOURCE_TYPE_UDF);
|
|
|
- createFolder(tenantCode+ FOLDER_SEPARATOR +RESOURCE_TYPE_FILE);
|
|
|
+ getInstance().mkdir(tenantCode, getS3ResDir(tenantCode));
|
|
|
+ getInstance().mkdir(tenantCode, getS3UdfDir(tenantCode));
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public String getResDir(String tenantCode) {
|
|
|
- return tenantCode+ FOLDER_SEPARATOR +RESOURCE_TYPE_FILE+FOLDER_SEPARATOR;
|
|
|
+ return getS3ResDir(tenantCode) + FOLDER_SEPARATOR;
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public String getUdfDir(String tenantCode) {
|
|
|
- return tenantCode+ FOLDER_SEPARATOR +RESOURCE_TYPE_UDF+FOLDER_SEPARATOR;
|
|
|
+ return getS3UdfDir(tenantCode) + FOLDER_SEPARATOR;
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public boolean mkdir(String tenantCode, String path) throws IOException {
|
|
|
- createFolder(path);
|
|
|
- return true;
|
|
|
+ String objectName = path + FOLDER_SEPARATOR;
|
|
|
+ if (!s3Client.doesObjectExist(BUCKET_NAME, objectName)) {
|
|
|
+ ObjectMetadata metadata = new ObjectMetadata();
|
|
|
+ metadata.setContentLength(0);
|
|
|
+ InputStream emptyContent = new ByteArrayInputStream(new byte[0]);
|
|
|
+ PutObjectRequest putObjectRequest = new PutObjectRequest(BUCKET_NAME, objectName, emptyContent, metadata);
|
|
|
+ s3Client.putObject(putObjectRequest);
|
|
|
+ }
|
|
|
+ return true;
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
@@ -150,21 +163,22 @@ public class S3Utils implements Closeable, StorageOperate {
|
|
|
if (fileName.startsWith(FOLDER_SEPARATOR)) {
|
|
|
fileName = fileName.replaceFirst(FOLDER_SEPARATOR, "");
|
|
|
}
|
|
|
- return String.format(FORMAT_S_S, tenantCode+FOLDER_SEPARATOR+RESOURCE_TYPE_FILE, fileName);
|
|
|
+ return String.format(FORMAT_S_S, getS3ResDir(tenantCode), fileName);
|
|
|
}
|
|
|
+
|
|
|
@Override
|
|
|
public String getFileName(ResourceType resourceType, String tenantCode, String fileName) {
|
|
|
if (fileName.startsWith(FOLDER_SEPARATOR)) {
|
|
|
fileName = fileName.replaceFirst(FOLDER_SEPARATOR, "");
|
|
|
}
|
|
|
- return getDir(resourceType, tenantCode)+fileName;
|
|
|
+ return getDir(resourceType, tenantCode) + fileName;
|
|
|
}
|
|
|
|
|
|
@Override
|
|
|
public void download(String tenantCode, String srcFilePath, String dstFile, boolean deleteSource, boolean overwrite) throws IOException {
|
|
|
S3Object o = s3Client.getObject(BUCKET_NAME, srcFilePath);
|
|
|
try (S3ObjectInputStream s3is = o.getObjectContent();
|
|
|
- FileOutputStream fos = new FileOutputStream(new File(dstFile))) {
|
|
|
+ FileOutputStream fos = new FileOutputStream(dstFile)) {
|
|
|
byte[] readBuf = new byte[1024];
|
|
|
int readLen = 0;
|
|
|
while ((readLen = s3is.read(readBuf)) > 0) {
|
|
|
@@ -210,7 +224,7 @@ public class S3Utils implements Closeable, StorageOperate {
|
|
|
case FILE:
|
|
|
return getResDir(tenantCode);
|
|
|
default:
|
|
|
- return tenantCode+ FOLDER_SEPARATOR ;
|
|
|
+ return "";
|
|
|
}
|
|
|
|
|
|
}
|
|
|
@@ -221,33 +235,21 @@ public class S3Utils implements Closeable, StorageOperate {
|
|
|
s3Client.putObject(BUCKET_NAME, dstPath, new File(srcFile));
|
|
|
return true;
|
|
|
} catch (AmazonServiceException e) {
|
|
|
- logger.error("upload failed,the bucketName is {},the dstPath is {}", BUCKET_NAME, tenantCode+ FOLDER_SEPARATOR +dstPath);
|
|
|
+ logger.error("upload failed,the bucketName is {},the filePath is {}", BUCKET_NAME, dstPath);
|
|
|
return false;
|
|
|
}
|
|
|
}
|
|
|
|
|
|
-
|
|
|
@Override
|
|
|
- public List<String> vimFile(String tenantCode,String filePath, int skipLineNums, int limit) throws IOException {
|
|
|
+ public List<String> vimFile(String tenantCode, String filePath, int skipLineNums, int limit) throws IOException {
|
|
|
if (StringUtils.isBlank(filePath)) {
|
|
|
logger.error("file path:{} is blank", filePath);
|
|
|
return Collections.emptyList();
|
|
|
}
|
|
|
- S3Object s3Object=s3Client.getObject(BUCKET_NAME,filePath);
|
|
|
- try(BufferedReader bufferedReader=new BufferedReader(new InputStreamReader(s3Object.getObjectContent()))){
|
|
|
- Stream<String> stream = bufferedReader.lines().skip(skipLineNums).limit(limit);
|
|
|
- return stream.collect(Collectors.toList());
|
|
|
- }
|
|
|
- }
|
|
|
-
|
|
|
- private void
|
|
|
- createFolder( String folderName) {
|
|
|
- if (!s3Client.doesObjectExist(BUCKET_NAME, folderName + FOLDER_SEPARATOR)) {
|
|
|
- ObjectMetadata metadata = new ObjectMetadata();
|
|
|
- metadata.setContentLength(0);
|
|
|
- InputStream emptyContent = new ByteArrayInputStream(new byte[0]);
|
|
|
- PutObjectRequest putObjectRequest = new PutObjectRequest(BUCKET_NAME, folderName + FOLDER_SEPARATOR, emptyContent, metadata);
|
|
|
- s3Client.putObject(putObjectRequest);
|
|
|
+ S3Object s3Object = s3Client.getObject(BUCKET_NAME, filePath);
|
|
|
+ try (BufferedReader bufferedReader = new BufferedReader(new InputStreamReader(s3Object.getObjectContent()))) {
|
|
|
+ Stream<String> stream = bufferedReader.lines().skip(skipLineNums).limit(limit);
|
|
|
+ return stream.collect(Collectors.toList());
|
|
|
}
|
|
|
}
|
|
|
|
|
|
@@ -256,6 +258,47 @@ public class S3Utils implements Closeable, StorageOperate {
|
|
|
deleteTenantCode(tenantCode);
|
|
|
}
|
|
|
|
|
|
+ /**
|
|
|
+ * S3 resource dir
|
|
|
+ *
|
|
|
+ * @param tenantCode tenant code
|
|
|
+ * @return S3 resource dir
|
|
|
+ */
|
|
|
+ public static String getS3ResDir(String tenantCode) {
|
|
|
+ return String.format("%s/" + RESOURCE_TYPE_FILE, getS3TenantDir(tenantCode));
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * S3 udf dir
|
|
|
+ *
|
|
|
+ * @param tenantCode tenant code
|
|
|
+ * @return get udf dir on S3
|
|
|
+ */
|
|
|
+ public static String getS3UdfDir(String tenantCode) {
|
|
|
+ return String.format("%s/" + RESOURCE_TYPE_UDF, getS3TenantDir(tenantCode));
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * @param tenantCode tenant code
|
|
|
+ * @return file directory of tenants on S3
|
|
|
+ */
|
|
|
+ public static String getS3TenantDir(String tenantCode) {
|
|
|
+ return String.format(FORMAT_S_S, getS3DataBasePath(), tenantCode);
|
|
|
+ }
|
|
|
+
|
|
|
+ /**
|
|
|
+ * get data S3 path
|
|
|
+ *
|
|
|
+ * @return data S3 path
|
|
|
+ */
|
|
|
+ public static String getS3DataBasePath() {
|
|
|
+ if (FOLDER_SEPARATOR.equals(RESOURCE_UPLOAD_PATH)) {
|
|
|
+ return "";
|
|
|
+ } else {
|
|
|
+ return RESOURCE_UPLOAD_PATH.replaceFirst(FOLDER_SEPARATOR, "");
|
|
|
+ }
|
|
|
+ }
|
|
|
+
|
|
|
private void deleteTenantCode(String tenantCode) {
|
|
|
deleteDirectory(getResDir(tenantCode));
|
|
|
deleteDirectory(getUdfDir(tenantCode));
|
|
|
@@ -264,25 +307,27 @@ public class S3Utils implements Closeable, StorageOperate {
|
|
|
/**
|
|
|
* xxx untest
|
|
|
* upload local directory to S3
|
|
|
+ *
|
|
|
* @param tenantCode
|
|
|
* @param keyPrefix the name of directory
|
|
|
* @param strPath
|
|
|
*/
|
|
|
private void uploadDirectory(String tenantCode, String keyPrefix, String strPath) {
|
|
|
- s3Client.putObject(BUCKET_NAME, tenantCode+ FOLDER_SEPARATOR +keyPrefix, new File(strPath));
|
|
|
+ s3Client.putObject(BUCKET_NAME, tenantCode + FOLDER_SEPARATOR + keyPrefix, new File(strPath));
|
|
|
}
|
|
|
|
|
|
|
|
|
/**
|
|
|
* xxx untest
|
|
|
* download S3 Directory to local
|
|
|
+ *
|
|
|
* @param tenantCode
|
|
|
* @param keyPrefix the name of directory
|
|
|
* @param srcPath
|
|
|
*/
|
|
|
- private void downloadDirectory(String tenantCode, String keyPrefix, String srcPath){
|
|
|
- TransferManager tm= TransferManagerBuilder.standard().withS3Client(s3Client).build();
|
|
|
- try{
|
|
|
+ private void downloadDirectory(String tenantCode, String keyPrefix, String srcPath) {
|
|
|
+ TransferManager tm = TransferManagerBuilder.standard().withS3Client(s3Client).build();
|
|
|
+ try {
|
|
|
MultipleFileDownload download = tm.downloadDirectory(BUCKET_NAME, tenantCode + FOLDER_SEPARATOR + keyPrefix, new File(srcPath));
|
|
|
download.waitForCompletion();
|
|
|
} catch (AmazonS3Exception | InterruptedException e) {
|
|
|
@@ -293,15 +338,26 @@ public class S3Utils implements Closeable, StorageOperate {
|
|
|
}
|
|
|
}
|
|
|
|
|
|
- public void checkBucketNameIfNotPresent(String bucketName) {
|
|
|
- if (!s3Client.doesBucketExistV2(bucketName)) {
|
|
|
- logger.info("the current regionName is {}", s3Client.getRegionName());
|
|
|
- s3Client.createBucket(bucketName);
|
|
|
+ public void checkBucketNameExists(String bucketName) {
|
|
|
+ if (StringUtils.isBlank(bucketName)) {
|
|
|
+ throw new IllegalArgumentException("resource.aws.s3.bucket.name is blank");
|
|
|
}
|
|
|
+
|
|
|
+ Bucket existsBucket = s3Client.listBuckets()
|
|
|
+ .stream()
|
|
|
+ .filter(
|
|
|
+ bucket -> bucket.getName().equals(bucketName)
|
|
|
+ )
|
|
|
+ .findFirst()
|
|
|
+ .orElseThrow(() -> {
|
|
|
+ return new IllegalArgumentException("bucketName: " + bucketName + " is not exists, you need to create them by yourself");
|
|
|
+ });
|
|
|
+
|
|
|
+ logger.info("bucketName: {} has been found, the current regionName is {}", existsBucket.getName(), s3Client.getRegionName());
|
|
|
}
|
|
|
|
|
|
- /*
|
|
|
- only delete the object of directory ,it`s better to delete the files in it -r
|
|
|
+ /**
|
|
|
+ * only delete the object of directory ,it`s better to delete the files in it -r
|
|
|
*/
|
|
|
private void deleteDirectory(String directoryName) {
|
|
|
if (s3Client.doesObjectExist(BUCKET_NAME, directoryName)) {
|
guodong