xxq250
/
homework-jianmu
1
0
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

cos/put: new arg withcp to control resumable uploading

This commit is contained in:
Minglei Jin 2023-11-30 17:20:21 +08:00
parent deec1c482d
commit 49ecb1093f
4 changed files with 18 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
include/common/cos.h
View File

@ -34,7 +34,7 @@ extern int32_t tsS3UploadDelaySec;
int32_t s3Init(); int32_t s3Init();
void s3CleanUp(); void s3CleanUp();
int32_t s3PutObjectFromFile(const char *file, const char *object); int32_t s3PutObjectFromFile(const char *file, const char *object);
int32_t s3PutObjectFromFile2(const char *file, const char *object); int32_t s3PutObjectFromFile2(const char *file, const char *object, int8_t withcp);
void s3DeleteObjectsByPrefix(const char *prefix); void s3DeleteObjectsByPrefix(const char *prefix);
void s3DeleteObjects(const char *object_name[], int nobject); void s3DeleteObjects(const char *object_name[], int nobject);
bool s3Exists(const char *object_name); bool s3Exists(const char *object_name);

16
source/common/src/cos.c
View File

@ -580,13 +580,18 @@ clean:
return code; return code;
} }
static int32_t s3PutObjectFromFileWithCp(S3BucketContext *bucket_context, char const *object_name,
int64_t contentLength, S3PutProperties *put_prop,
put_object_callback_data *data) {
/*
static int32_t s3PutObjectFromFileWithCp(const char *file, int64_t size, int32_t lmtime, const char *object) { static int32_t s3PutObjectFromFileWithCp(const char *file, int64_t size, int32_t lmtime, const char *object) {
*/
int32_t code = 0; int32_t code = 0;
return code; return code;
} }
int32_t s3PutObjectFromFile2(const char *file, const char *object_name) { int32_t s3PutObjectFromFile2(const char *file, const char *object_name, int8_t withcp) {
int32_t code = 0; int32_t code = 0;
int32_t lmtime = 0; int32_t lmtime = 0;
const char *filename = 0; const char *filename = 0;
@ -626,9 +631,13 @@ int32_t s3PutObjectFromFile2(const char *file, const char *object_name) {
if (contentLength <= MULTIPART_CHUNK_SIZE) { if (contentLength <= MULTIPART_CHUNK_SIZE) {
code = s3PutObjectFromFileSimple(&bucketContext, object_name, contentLength, &putProperties, &data); code = s3PutObjectFromFileSimple(&bucketContext, object_name, contentLength, &putProperties, &data);
} else {
if (withcp) {
code = s3PutObjectFromFileWithCp(&bucketContext, object_name, contentLength, &putProperties, &data);
} else { } else {
code = s3PutObjectFromFileWithoutCp(&bucketContext, object_name, contentLength, &putProperties, &data); code = s3PutObjectFromFileWithoutCp(&bucketContext, object_name, contentLength, &putProperties, &data);
} }
}
if (data.infileFD) { if (data.infileFD) {
taosCloseFile(&data.infileFD); taosCloseFile(&data.infileFD);
@ -981,7 +990,7 @@ int32_t s3PutObjectFromFile(const char *file_str, const char *object_str) {
return code; return code;
} }
int32_t s3PutObjectFromFile2(const char *file_str, const char *object_str) { int32_t s3PutObjectFromFile2(const char *file_str, const char *object_str, int8_t withcp) {
int32_t code = 0; int32_t code = 0;
cos_pool_t *p = NULL; cos_pool_t *p = NULL;
int is_cname = 0; int is_cname = 0;
@ -993,6 +1002,7 @@ int32_t s3PutObjectFromFile2(const char *file_str, const char *object_str) {
cos_table_t *headers = NULL; cos_table_t *headers = NULL;
cos_resumable_clt_params_t *clt_params = NULL; cos_resumable_clt_params_t *clt_params = NULL;
(void)withcp;
cos_pool_create(&p, NULL); cos_pool_create(&p, NULL);
options = cos_request_options_create(p); options = cos_request_options_create(p);
s3InitRequestOptions(options, is_cname); s3InitRequestOptions(options, is_cname);
@ -1342,7 +1352,7 @@ long s3Size(const char *object_name) {
int32_t s3Init() { return 0; } int32_t s3Init() { return 0; }
void s3CleanUp() {} void s3CleanUp() {}
int32_t s3PutObjectFromFile(const char *file, const char *object) { return 0; } int32_t s3PutObjectFromFile(const char *file, const char *object) { return 0; }
int32_t s3PutObjectFromFile2(const char *file, const char *object) { return 0; } int32_t s3PutObjectFromFile2(const char *file, const char *object, int8_t withcp) { return 0; }
void s3DeleteObjectsByPrefix(const char *prefix) {} void s3DeleteObjectsByPrefix(const char *prefix) {}
void s3DeleteObjects(const char *object_name[], int nobject) {} void s3DeleteObjects(const char *object_name[], int nobject) {}
bool s3Exists(const char *object_name) { return false; } bool s3Exists(const char *object_name) { return false; }

2
source/dnode/vnode/src/tsdb/tsdbRetention.c
View File

@ -112,7 +112,7 @@ static int32_t tsdbCopyFileS3(SRTNer *rtner, const STFileObj *from, const STFile
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
char *object_name = taosDirEntryBaseName(fname); char *object_name = taosDirEntryBaseName(fname);
code = s3PutObjectFromFile2(from->fname, object_name); code = s3PutObjectFromFile2(from->fname, object_name, 1);
TSDB_CHECK_CODE(code, lino, _exit); TSDB_CHECK_CODE(code, lino, _exit);
taosCloseFile(&fdFrom); taosCloseFile(&fdFrom);

2
source/libs/stream/src/streamCheckpoint.c
View File

@ -515,7 +515,7 @@ static int uploadCheckpointToS3(char* id, char* path) {
char object[PATH_MAX] = {0}; char object[PATH_MAX] = {0};
snprintf(object, sizeof(object), "%s%s%s", id, TD_DIRSEP, name); snprintf(object, sizeof(object), "%s%s%s", id, TD_DIRSEP, name);
if (s3PutObjectFromFile2(filename, object) != 0) { if (s3PutObjectFromFile2(filename, object, 0) != 0) {
taosCloseDir(&pDir); taosCloseDir(&pDir);
return -1; return -1;
} }