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

fix(stream): fix error in downloading remote backup checkpoint data.

This commit is contained in:
Haojun Liao 2024-05-11 16:19:11 +08:00
parent f8d8061260
commit d214dd5cdd
2 changed files with 26 additions and 23 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
source/common/src/rsync.c
View File

@ -196,10 +196,14 @@ int32_t uploadRsync(const char* id, const char* path) {
}
int32_t downloadRsync(const char* id, const char* path) {
int64_t st = taosGetTimestampMs();
uDebug("[rsync] %s start to sync data from remote to local:%s", id, path);
#ifdef WINDOWS
char pathTransform[PATH_MAX] = {0};
changeDirFromWindowsToLinux(path, pathTransform);
#endif
char command[PATH_MAX] = {0};
snprintf(command, PATH_MAX, "rsync -av --timeout=10 --bwlimit=100000 rsync://%s/checkpoint/%s/ %s",
tsSnodeAddress, id,
@ -211,13 +215,15 @@ int32_t downloadRsync(const char* id, const char* path) {
);
int32_t code = execCommand(command);
int32_t el = taosGetTimestampMs() - st;
if (code != 0) {
uError("[rsync] download checkpoint data failed, code:%d," ERRNO_ERR_FORMAT, code, ERRNO_ERR_DATA);
return -1;
uError("[rsync] %s download checkpoint data:%s failed, code:%d," ERRNO_ERR_FORMAT, id, path, code, ERRNO_ERR_DATA);
} else {
uDebug("[rsync] %s download checkpoint data:%s successfully, elapsed time:%dms", id, path, el);
}
uDebug("[rsync] download checkpoint data:%s successfully", id);
return 0;
return code;
}
int32_t deleteRsync(const char* id) {

33
source/libs/stream/src/streamBackendRocksdb.c
View File

@ -325,28 +325,27 @@ int32_t remoteChkp_validAndCvtMeta(char* path, SArray* list, int64_t chkpId) {
return complete == 1 ? 0 : -1;
}
int32_t rebuildFromRemoteChkp_rsync(char* key, char* chkpPath, int64_t checkpointId, char* defaultPath) {
// impl later
int32_t rebuildFromRemoteChkp_rsync(char* key, char* chkptPath, int64_t checkpointId, char* defaultPath) {
int32_t code = 0;
if (taosIsDir(chkpPath)) {
taosRemoveDir(chkpPath);
if (taosIsDir(chkptPath)) {
taosRemoveDir(chkptPath);
stDebug("remove local checkpoint data dir:%s succ", chkptPath);
}
if (taosIsDir(defaultPath)) {
taosRemoveDir(defaultPath);
taosMulMkDir(defaultPath);
stDebug("clear local backend dir:%s succ", defaultPath);
}
code = streamTaskDownloadCheckpointData(key, chkpPath);
code = streamTaskDownloadCheckpointData(key, chkptPath);
if (code != 0) {
stError("failed to download checkpoint data:%s", key);
return code;
}
stDebug("download backup checkpoint data into:%s, checkpointId:%" PRId64 ", %s", chkpPath, checkpointId, key);
code = backendCopyFiles(chkpPath, defaultPath);
return code;
stDebug("download remote checkpoint data for checkpointId:%" PRId64 ", %s", checkpointId, key);
return backendCopyFiles(chkptPath, defaultPath);
}
int32_t rebuildFromRemoteChkp_s3(char* key, char* chkpPath, int64_t chkpId, char* defaultPath) {
@ -485,14 +484,13 @@ int32_t backendCopyFiles(const char* src, const char* dst) {
return backendFileCopyFilesImpl(src, dst);
}
static int32_t rebuildFromLocalCheckpoint(char* pTaskIdStr, const char* checkpointPath, int64_t chkpId, const char* defaultPath) {
static int32_t rebuildFromLocalCheckpoint(char* pTaskIdStr, const char* checkpointPath, int64_t chkptId, const char* defaultPath) {
int32_t code = 0;
if (taosIsDir(defaultPath)) {
taosRemoveDir(defaultPath);
taosMkDir(defaultPath);
stInfo("clear task backend path:%s, done", defaultPath);
stInfo("clear task backend dir:%s, done", defaultPath);
}
if (taosIsDir(checkpointPath) && isValidCheckpoint(checkpointPath)) {
@ -506,11 +504,12 @@ static int32_t rebuildFromLocalCheckpoint(char* pTaskIdStr, const char* checkpoi
pTaskIdStr, checkpointPath, tstrerror(TAOS_SYSTEM_ERROR(errno)), defaultPath);
code = TSDB_CODE_SUCCESS;
} else {
stInfo("%s start to restart stream backend at checkpoint path: %s", pTaskIdStr, checkpointPath);
stInfo("%s copy checkpoint data from:%s to:%s succ, try to start stream backend", pTaskIdStr, checkpointPath,
defaultPath);
}
} else {
code = TSDB_CODE_FAILED;
stError("%s not valid checkpoint path/data in:%s", pTaskIdStr, checkpointPath);
stError("%s no valid checkpoint data for checkpointId:%" PRId64 " in %s", pTaskIdStr, chkptId, checkpointPath);
}
return code;
@ -522,7 +521,6 @@ int32_t rebuildFromlocalDefault(char* key, char* chkpPath, int64_t chkpId, char*
}
int32_t rebuildDirFormCheckpoint(const char* path, char* key, int64_t chkptId, char** dbPrefixPath, char** dbPath) {
// impl later
int32_t code = 0;
char* prefixPath = taosMemoryCalloc(1, strlen(path) + 128);
@ -538,8 +536,7 @@ int32_t rebuildDirFormCheckpoint(const char* path, char* key, int64_t chkptId, c
if (!taosIsDir(defaultPath)) {
taosMulMkDir(defaultPath);
}
stDebug("prepare local dir:%s, checkpointId:%" PRId64 ", key:%s succ", defaultPath, chkptId, key);
stDebug("local default dir:%s, checkpointId:%" PRId64 ", key:%s succ", defaultPath, chkptId, key);
char* chkptPath = taosMemoryCalloc(1, strlen(path) + 256);
if (chkptId != 0) {