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

fix(stream): update the remote checkpoint directory in snode.

This commit is contained in:
Haojun Liao 2024-06-25 11:42:54 +08:00
parent 6b0b65527f
commit 8cda12bc18
1 changed files with 49 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

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

@ -170,7 +170,7 @@ int32_t uploadByRsync(const char* id, const char* path) {
if (path[strlen(path) - 1] != '/') {
#endif
snprintf(command, PATH_MAX,
"rsync -av --debug=all --log-file=%s/rsynclog --delete --timeout=10 --bwlimit=100000 %s/ "
"rsync -av --debug=all --log-file=%s/rsynclog --delete --timeout=10 --bwlimit=100000 --exclude=\"*\" %s/ "
"rsync://%s/checkpoint/%s/",
tsLogDir,
#ifdef WINDOWS
@ -182,7 +182,7 @@ int32_t uploadByRsync(const char* id, const char* path) {
tsSnodeAddress, id);
} else {
snprintf(command, PATH_MAX,
"rsync -av --debug=all --log-file=%s/rsynclog --delete --timeout=10 --bwlimit=100000 %s "
"rsync -av --debug=all --log-file=%s/rsynclog --delete --timeout=10 --bwlimit=100000 --exclude=\"*\" %s "
"rsync://%s/checkpoint/%s/",
tsLogDir,
#ifdef WINDOWS
@ -194,7 +194,51 @@ int32_t uploadByRsync(const char* id, const char* path) {
tsSnodeAddress, id);
}
// prepare the data directory
int32_t code = execCommand(command);
if (code != 0) {
uError("[rsync] s-task:%s prepare checkpoint data in %s to %s failed, code:%d," ERRNO_ERR_FORMAT, id, path,
tsSnodeAddress, code, ERRNO_ERR_DATA);
} else {
int64_t el = (taosGetTimestampMs() - st);
uDebug("[rsync] s-task:%s prepare checkpoint data in:%s to %s successfully, elapsed time:%" PRId64 "ms", id, path,
tsSnodeAddress, el);
}
#ifdef WINDOWS
char pathTransform[PATH_MAX] = {0};
changeDirFromWindowsToLinux(path, pathTransform);
if (pathTransform[strlen(pathTransform) - 1] != '/') {
#else
if (path[strlen(path) - 1] != '/') {
#endif
snprintf(command, PATH_MAX,
"rsync -av --debug=all --log-file=%s/rsynclog --delete --timeout=10 --bwlimit=100000 %s/ "
"rsync://%s/checkpoint/%s/data/",
tsLogDir,
#ifdef WINDOWS
pathTransform
#else
path
#endif
,
tsSnodeAddress, id);
} else {
snprintf(command, PATH_MAX,
"rsync -av --debug=all --log-file=%s/rsynclog --delete --timeout=10 --bwlimit=100000 %s "
"rsync://%s/checkpoint/%s/data/",
tsLogDir,
#ifdef WINDOWS
pathTransform
#else
path
#endif
,
tsSnodeAddress, id);
}
code = execCommand(command);
if (code != 0) {
uError("[rsync] s-task:%s upload checkpoint data in %s to %s failed, code:%d," ERRNO_ERR_FORMAT, id, path,
tsSnodeAddress, code, ERRNO_ERR_DATA);
@ -210,7 +254,7 @@ int32_t uploadByRsync(const char* id, const char* path) {
// abort from retry if quit
int32_t downloadRsync(const char* id, const char* path) {
int64_t st = taosGetTimestampMs();
int32_t MAX_RETRY = 60;
int32_t MAX_RETRY = 10;
int32_t times = 0;
int32_t code = 0;
@ -221,7 +265,7 @@ int32_t downloadRsync(const char* id, const char* path) {
char command[PATH_MAX] = {0};
snprintf(command, PATH_MAX,
"rsync -av --debug=all --log-file=%s/rsynclog --timeout=10 --bwlimit=100000 rsync://%s/checkpoint/%s/ %s",
"rsync -av --debug=all --log-file=%s/rsynclog --timeout=10 --bwlimit=100000 rsync://%s/checkpoint/%s/data/ %s",
tsLogDir, tsSnodeAddress, id,
#ifdef WINDOWS
pathTransform
@ -258,7 +302,7 @@ int32_t deleteRsync(const char* id) {
char command[PATH_MAX] = {0};
snprintf(command, PATH_MAX,
"rsync -av --debug=all --log-file=%s/rsynclog --delete --timeout=10 %s rsync://%s/checkpoint/%s/", tsLogDir,
"rsync -av --debug=all --log-file=%s/rsynclog --delete --timeout=10 %s rsync://%s/checkpoint/%s/data/", tsLogDir,
tmp, tsSnodeAddress, id);
code = execCommand(command);