database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[feature](backup) add property to remove snapshot before creating repo (#25847)

Browse Source 
Doris is not responsible for managing snapshots, but it needs to clear all
snapshots before doing backup/restore regression testing, so a property is
added to indicate that existing snapshots need to be cleared when creating a
repo.

In addition, a regression test case for backup/restore has been added.
This commit is contained in:
walter
2023-10-27 21:03:26 +08:00
committed by GitHub
parent c715facafa
commit 365fdd2f4d
11 changed files with 440 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

12
fe/fe-core/src/main/java/org/apache/doris/analysis/CreateRepositoryStmt.java
View File

@ -28,6 +28,8 @@ import org.apache.doris.qe.ConnectContext;
import java.util.Map;
public class CreateRepositoryStmt extends DdlStmt {
public static String PROP_DELETE_IF_EXISTS = "delete_if_exists";
private boolean isReadOnly;
private String name;
private StorageBackend storage;
@ -71,6 +73,16 @@ public class CreateRepositoryStmt extends DdlStmt {
ErrorReport.reportAnalysisException(ErrorCode.ERR_SPECIFIC_ACCESS_DENIED_ERROR, "ADMIN");
}
FeNameFormat.checkCommonName("repository", name);
// check delete_if_exists, this property will be used by Repository.initRepository.
Map<String, String> properties = getProperties();
String deleteIfExistsStr = properties.get(PROP_DELETE_IF_EXISTS);
if (deleteIfExistsStr != null) {
if (!deleteIfExistsStr.equalsIgnoreCase("true") && !deleteIfExistsStr.equalsIgnoreCase("false")) {
ErrorReport.reportAnalysisException(ErrorCode.ERR_COMMON_ERROR,
"'" + PROP_DELETE_IF_EXISTS + "' in properties, you should set it false or true");
}
}
}
@Override

24
fe/fe-core/src/main/java/org/apache/doris/backup/Repository.java
View File

@ -17,6 +17,7 @@
package org.apache.doris.backup;
import org.apache.doris.analysis.CreateRepositoryStmt;
import org.apache.doris.analysis.StorageBackend;
import org.apache.doris.backup.Status.ErrCode;
import org.apache.doris.catalog.Env;
@ -215,6 +216,27 @@ public class Repository implements Writable {
if (FeConstants.runningUnitTest) {
return Status.OK;
}
// A temporary solution is to delete all stale snapshots before creating an S3 repository
// so that we can add regression tests about backup/restore.
//
// TODO: support hdfs/brokers
if (fileSystem instanceof S3FileSystem) {
String deleteStaledSnapshots = fileSystem.getProperties()
.getOrDefault(CreateRepositoryStmt.PROP_DELETE_IF_EXISTS, "false");
if (deleteStaledSnapshots.equalsIgnoreCase("true")) {
// delete with prefix:
// eg. __palo_repository_repo_name/
String snapshotPrefix = Joiner.on(PATH_DELIMITER).join(location, joinPrefix(PREFIX_REPO, name));
LOG.info("property {} is set, delete snapshots with prefix: {}",
CreateRepositoryStmt.PROP_DELETE_IF_EXISTS, snapshotPrefix);
Status st = ((S3FileSystem) fileSystem).deleteDirectory(snapshotPrefix);
if (!st.ok()) {
return st;
}
}
}
String repoInfoFilePath = assembleRepoInfoFilePath();
// check if the repo is already exist in remote
List<RemoteFile> remoteFiles = Lists.newArrayList();
@ -245,8 +267,8 @@ public class Repository implements Writable {
return new Status(ErrCode.COMMON_ERROR,
"failed to parse create time of repository: " + root.get("create_time"));
}
return Status.OK;
return Status.OK;
} catch (IOException e) {
return new Status(ErrCode.COMMON_ERROR, "failed to read repo info file: " + e.getMessage());
} finally {

2
fe/fe-core/src/main/java/org/apache/doris/fs/obj/ObjStorage.java
View File

@ -44,6 +44,8 @@ public interface ObjStorage<C> {
Status deleteObject(String remotePath);
Status deleteObjects(String remotePath);
Status copyObject(String origFilePath, String destFilePath);
RemoteObjects listObjects(String remotePath, String continuationToken) throws DdlException;

74
fe/fe-core/src/main/java/org/apache/doris/fs/obj/S3ObjStorage.java
View File

@ -37,14 +37,18 @@ import software.amazon.awssdk.core.sync.RequestBody;
import software.amazon.awssdk.services.s3.S3Client;
import software.amazon.awssdk.services.s3.model.CopyObjectRequest;
import software.amazon.awssdk.services.s3.model.CopyObjectResponse;
import software.amazon.awssdk.services.s3.model.Delete;
import software.amazon.awssdk.services.s3.model.DeleteObjectRequest;
import software.amazon.awssdk.services.s3.model.DeleteObjectResponse;
import software.amazon.awssdk.services.s3.model.DeleteObjectsRequest;
import software.amazon.awssdk.services.s3.model.DeleteObjectsResponse;
import software.amazon.awssdk.services.s3.model.GetObjectRequest;
import software.amazon.awssdk.services.s3.model.GetObjectResponse;
import software.amazon.awssdk.services.s3.model.HeadObjectRequest;
import software.amazon.awssdk.services.s3.model.HeadObjectResponse;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Request;
import software.amazon.awssdk.services.s3.model.ListObjectsV2Response;
import software.amazon.awssdk.services.s3.model.ObjectIdentifier;
import software.amazon.awssdk.services.s3.model.PutObjectRequest;
import software.amazon.awssdk.services.s3.model.PutObjectResponse;
import software.amazon.awssdk.services.s3.model.S3Exception;
@ -56,6 +60,7 @@ import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.TreeMap;
import java.util.stream.Collectors;
public class S3ObjStorage implements ObjStorage<S3Client> {
private static final Logger LOG = LogManager.getLogger(S3ObjStorage.class);
@ -223,6 +228,52 @@ public class S3ObjStorage implements ObjStorage<S3Client> {
}
}
@Override
public Status deleteObjects(String absolutePath) {
try {
S3URI baseUri = S3URI.create(absolutePath, forceHostedStyle);
String continuationToken = "";
boolean isTruncated = false;
long totalObjects = 0;
do {
RemoteObjects objects = listObjects(absolutePath, continuationToken);
List<RemoteObject> objectList = objects.getObjectList();
if (!objectList.isEmpty()) {
Delete delete = Delete.builder()
.objects(objectList.stream()
.map(RemoteObject::getKey)
.map(k -> ObjectIdentifier.builder().key(k).build())
.collect(Collectors.toList()))
.build();
DeleteObjectsRequest req = DeleteObjectsRequest.builder()
.bucket(baseUri.getBucket())
.delete(delete)
.build();
DeleteObjectsResponse resp = getClient(baseUri.getVirtualBucket()).deleteObjects(req);
if (resp.errors().size() > 0) {
LOG.warn("{} errors returned while deleting {} objects for dir {}",
resp.errors().size(), objectList.size(), absolutePath);
}
LOG.info("{} of {} objects deleted for dir {}",
resp.deleted().size(), objectList.size(), absolutePath);
totalObjects += objectList.size();
}
isTruncated = objects.isTruncated();
continuationToken = objects.getContinuationToken();
} while (isTruncated);
LOG.info("total delete {} objects for dir {}", totalObjects, absolutePath);
return Status.OK;
} catch (DdlException e) {
return new Status(Status.ErrCode.COMMON_ERROR, "list objects for delete objects failed: " + e.getMessage());
} catch (Exception e) {
LOG.warn("delete objects {} failed, force visual host style {}", absolutePath, e, forceHostedStyle);
return new Status(Status.ErrCode.COMMON_ERROR, "delete objects failed: " + e.getMessage());
}
}
@Override
public Status copyObject(String origFilePath, String destFilePath) {
try {
S3URI origUri = S3URI.create(origFilePath);
@ -249,9 +300,26 @@ public class S3ObjStorage implements ObjStorage<S3Client> {
public RemoteObjects listObjects(String absolutePath, String continuationToken) throws DdlException {
try {
S3URI uri = S3URI.create(absolutePath, forceHostedStyle);
String bucket = uri.getBucket();
String prefix = uri.getKey();
ListObjectsV2Request.Builder requestBuilder = ListObjectsV2Request.builder().bucket(uri.getBucket())
.prefix(normalizePrefix(prefix));
if (!StringUtils.isEmpty(uri.getVirtualBucket())) {
// Support s3 compatible service. The generated HTTP request for list objects likes:
//
// GET /<bucket-name>?list-type=2&prefix=<prefix>
prefix = bucket + "/" + prefix;
String endpoint = properties.get(S3Properties.ENDPOINT);
if (endpoint.contains("cos.")) {
bucket = "/";
} else if (endpoint.contains("oss-")) {
bucket = uri.getVirtualBucket();
} else if (endpoint.contains("obs.")) {
// FIXME: unlike cos and oss, the obs will report 'The specified key does not exist'.
throw new DdlException("obs does not support list objects via s3 sdk. path: " + absolutePath);
}
}
ListObjectsV2Request.Builder requestBuilder = ListObjectsV2Request.builder()
.bucket(bucket)
.prefix(normalizePrefix(prefix));
if (!StringUtils.isEmpty(continuationToken)) {
requestBuilder.continuationToken(continuationToken);
}
@ -263,7 +331,7 @@ public class S3ObjStorage implements ObjStorage<S3Client> {
}
return new RemoteObjects(remoteObjects, response.isTruncated(), response.nextContinuationToken());
} catch (Exception e) {
LOG.warn("Failed to list objects for S3", e);
LOG.warn("Failed to list objects for S3: {}", absolutePath, e);
throw new DdlException("Failed to list objects for S3, Error message: " + e.getMessage(), e);
}
}

4
fe/fe-core/src/main/java/org/apache/doris/fs/remote/S3FileSystem.java
View File

@ -107,5 +107,9 @@ public class S3FileSystem extends ObjFileSystem {
}
return Status.OK;
}
public Status deleteDirectory(String absolutePath) {
return ((S3ObjStorage) objStorage).deleteObjects(absolutePath);
}
}