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

fix:[TS-4921]refactor code

This commit is contained in:
wangmm0220 2024-07-05 14:03:12 +08:00
parent 20c8e3168c
commit 6c5acdfc4b
2 changed files with 40 additions and 47 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

7
include/libs/monitor/clientMonitor.h
View File

@ -38,6 +38,13 @@ typedef enum {
SLOW_LOG_READ_QUIT = 3, SLOW_LOG_READ_QUIT = 3,
} SLOW_LOG_QUEUE_TYPE; } SLOW_LOG_QUEUE_TYPE;
char* queueTypeStr[] = {
"SLOW_LOG_WRITE",
"SLOW_LOG_READ_RUNNING",
"SLOW_LOG_READ_BEGINNIG",
"SLOW_LOG_READ_QUIT"
};
#define SLOW_LOG_SEND_SIZE_MAX 1024*1024 #define SLOW_LOG_SEND_SIZE_MAX 1024*1024
typedef struct { typedef struct {

80
source/client/src/clientMonitor.c
View File

@ -454,6 +454,10 @@ static int64_t getFileSize(char* path){
} }
static int32_t sendSlowLog(int64_t clusterId, char* data, TdFilePtr pFile, int64_t offset, SLOW_LOG_QUEUE_TYPE type, char* fileName, void* pTransporter, SEpSet *epSet){ static int32_t sendSlowLog(int64_t clusterId, char* data, TdFilePtr pFile, int64_t offset, SLOW_LOG_QUEUE_TYPE type, char* fileName, void* pTransporter, SEpSet *epSet){
if (data == NULL){
taosMemoryFree(fileName);
return -1;
}
MonitorSlowLogData* pParam = taosMemoryMalloc(sizeof(MonitorSlowLogData)); MonitorSlowLogData* pParam = taosMemoryMalloc(sizeof(MonitorSlowLogData));
if(pParam == NULL){ if(pParam == NULL){
taosMemoryFree(data); taosMemoryFree(data);
@ -469,17 +473,26 @@ static int32_t sendSlowLog(int64_t clusterId, char* data, TdFilePtr pFile, int64
return sendReport(pTransporter, epSet, data, MONITOR_TYPE_SLOW_LOG, pParam); return sendReport(pTransporter, epSet, data, MONITOR_TYPE_SLOW_LOG, pParam);
} }
static void monitorSendSlowLogAtBeginning(int64_t clusterId, char** fileName, TdFilePtr pFile, int64_t offset, void* pTransporter, SEpSet *epSet){ static int32_t monitorReadSend(int64_t clusterId, TdFilePtr pFile, int64_t* offset, int64_t size, SLOW_LOG_QUEUE_TYPE type, char* fileName){
SAppInstInfo* pInst = getAppInstByClusterId(clusterId);
if(pInst == NULL){
tscError("failed to get app instance by clusterId:%" PRId64, clusterId);
return -1;
}
SEpSet ep = getEpSet_s(&pInst->mgmtEp);
char* data = readFile(pFile, offset, size);
return sendSlowLog(clusterId, data, (type == SLOW_LOG_READ_BEGINNIG ? pFile : NULL), *offset, type, fileName, pInst->pTransporter, &ep);
}
static void monitorSendSlowLogAtBeginning(int64_t clusterId, char** fileName, TdFilePtr pFile, int64_t offset){
int64_t size = getFileSize(*fileName); int64_t size = getFileSize(*fileName);
if(size <= offset){ if(size <= offset){
processFileInTheEnd(pFile, *fileName); processFileInTheEnd(pFile, *fileName);
tscDebug("[monitor] monitorSendSlowLogAtBeginning delete file:%s", *fileName); tscDebug("[monitor] monitorSendSlowLogAtBeginning delete file:%s", *fileName);
}else{ }else{
char* data = readFile(pFile, &offset, size); int32_t code = monitorReadSend(clusterId, pFile, &offset, size, SLOW_LOG_READ_BEGINNIG, *fileName);
if(data != NULL){ tscDebug("[monitor] monitorSendSlowLogAtBeginning send slow log clusterId:%"PRId64",ret:%d", clusterId, code);
sendSlowLog(clusterId, data, pFile, offset, SLOW_LOG_READ_BEGINNIG, *fileName, pTransporter, epSet); *fileName = NULL;
*fileName = NULL;
}
} }
} }
@ -500,16 +513,8 @@ static void monitorSendSlowLogAtRunning(int64_t clusterId){
tscDebug("[monitor] monitorSendSlowLogAtRunning truncate file to 0 file:%p", pClient->pFile); tscDebug("[monitor] monitorSendSlowLogAtRunning truncate file to 0 file:%p", pClient->pFile);
pClient->offset = 0; pClient->offset = 0;
}else{ }else{
SAppInstInfo* pInst = getAppInstByClusterId(clusterId); int32_t code = monitorReadSend(clusterId, pClient->pFile, &pClient->offset, size, SLOW_LOG_READ_RUNNING, NULL);
if(pInst == NULL){ tscDebug("[monitor] monitorSendSlowLogAtRunning send slow log clusterId:%"PRId64",ret:%d", clusterId, code);
tscError("failed to get app instance by clusterId:%" PRId64, clusterId);
return;
}
SEpSet ep = getEpSet_s(&pInst->mgmtEp);
char* data = readFile(pClient->pFile, &pClient->offset, size);
if(data != NULL){
sendSlowLog(clusterId, data, pClient->pFile, pClient->offset, SLOW_LOG_READ_RUNNING, NULL, pInst->pTransporter, &ep);
}
} }
} }
@ -531,15 +536,8 @@ static bool monitorSendSlowLogAtQuit(int64_t clusterId) {
return true; return true;
} }
}else{ }else{
SAppInstInfo* pInst = getAppInstByClusterId(clusterId); int32_t code = monitorReadSend(clusterId, pClient->pFile, &pClient->offset, size, SLOW_LOG_READ_QUIT, NULL);
if(pInst == NULL) { tscDebug("[monitor] monitorSendSlowLogAtQuit send slow log clusterId:%"PRId64",ret:%d", clusterId, code);
return true;
}
SEpSet ep = getEpSet_s(&pInst->mgmtEp);
char* data = readFile(pClient->pFile, &pClient->offset, size);
if(data != NULL){
sendSlowLog(clusterId, data, pClient->pFile, pClient->offset, SLOW_LOG_READ_QUIT, NULL, pInst->pTransporter, &ep);
}
} }
return false; return false;
} }
@ -556,13 +554,9 @@ static void monitorSendAllSlowLogAtQuit(){
pClient->pFile = NULL; pClient->pFile = NULL;
}else if(pClient->offset == 0){ }else if(pClient->offset == 0){
int64_t* clusterId = (int64_t*)taosHashGetKey(pIter, NULL); int64_t* clusterId = (int64_t*)taosHashGetKey(pIter, NULL);
SAppInstInfo* pInst = getAppInstByClusterId(*clusterId); int32_t code = monitorReadSend(*clusterId, pClient->pFile, &pClient->offset, size, SLOW_LOG_READ_QUIT, NULL);
if(pInst == NULL) { tscDebug("[monitor] monitorSendAllSlowLogAtQuit send slow log clusterId:%"PRId64",ret:%d", *clusterId, code);
continue; if (code == 0){
}
SEpSet ep = getEpSet_s(&pInst->mgmtEp);
char* data = readFile(pClient->pFile, &pClient->offset, size);
if(data != NULL && sendSlowLog(*clusterId, data, NULL, pClient->offset, SLOW_LOG_READ_QUIT, NULL, pInst->pTransporter, &ep) == 0){
quitCnt ++; quitCnt ++;
} }
} }
@ -610,11 +604,8 @@ static void monitorSendAllSlowLog(){
} }
continue; continue;
} }
SEpSet ep = getEpSet_s(&pInst->mgmtEp); int32_t code = monitorReadSend(*clusterId, pClient->pFile, &pClient->offset, size, SLOW_LOG_READ_RUNNING, NULL);
char* data = readFile(pClient->pFile, &pClient->offset, size); tscDebug("[monitor] monitorSendAllSlowLog send slow log clusterId:%"PRId64",ret:%d", *clusterId, code);
if(data != NULL){
sendSlowLog(*clusterId, data, NULL, pClient->offset, SLOW_LOG_READ_RUNNING, NULL, pInst->pTransporter, &ep);
}
} }
} }
} }
@ -627,7 +618,7 @@ static void monitorSendAllSlowLogFromTempDir(int64_t clusterId){
return; return;
} }
char namePrefix[PATH_MAX] = {0}; char namePrefix[PATH_MAX] = {0};
if (snprintf(namePrefix, sizeof(namePrefix), "%s%"PRIx64, TD_TMP_FILE_PREFIX, pInst->clusterId) < 0) { if (snprintf(namePrefix, sizeof(namePrefix), "%s%"PRIx64, TD_TMP_FILE_PREFIX, clusterId) < 0) {
tscError("failed to generate slow log file name prefix"); tscError("failed to generate slow log file name prefix");
return; return;
} }
@ -652,7 +643,7 @@ static void monitorSendAllSlowLogFromTempDir(int64_t clusterId){
if (strcmp(name, ".") == 0 || if (strcmp(name, ".") == 0 ||
strcmp(name, "..") == 0 || strcmp(name, "..") == 0 ||
strstr(name, namePrefix) == NULL) { strstr(name, namePrefix) == NULL) {
tscInfo("skip file:%s, for cluster id:%"PRIx64, name, pInst->clusterId); tscInfo("skip file:%s, for cluster id:%"PRIx64, name, clusterId);
continue; continue;
} }
@ -668,9 +659,8 @@ static void monitorSendAllSlowLogFromTempDir(int64_t clusterId){
taosCloseFile(&pFile); taosCloseFile(&pFile);
continue; continue;
} }
SEpSet ep = getEpSet_s(&pInst->mgmtEp);
char *tmp = taosStrdup(filename); char *tmp = taosStrdup(filename);
monitorSendSlowLogAtBeginning(pInst->clusterId, &tmp, pFile, 0, pInst->pTransporter, &ep); monitorSendSlowLogAtBeginning(clusterId, &tmp, pFile, 0);
taosMemoryFree(tmp); taosMemoryFree(tmp);
} }
@ -712,11 +702,7 @@ static void* monitorThreadFunc(void *param){
if (slowLogData != NULL) { if (slowLogData != NULL) {
if (slowLogData->type == SLOW_LOG_READ_BEGINNIG){ if (slowLogData->type == SLOW_LOG_READ_BEGINNIG){
if(slowLogData->pFile != NULL){ if(slowLogData->pFile != NULL){
SAppInstInfo* pInst = getAppInstByClusterId(slowLogData->clusterId); monitorSendSlowLogAtBeginning(slowLogData->clusterId, &(slowLogData->fileName), slowLogData->pFile, slowLogData->offset);
if(pInst != NULL) {
SEpSet ep = getEpSet_s(&pInst->mgmtEp);
monitorSendSlowLogAtBeginning(slowLogData->clusterId, &(slowLogData->fileName), slowLogData->pFile, slowLogData->offset, pInst->pTransporter, &ep);
}
}else{ }else{
monitorSendAllSlowLogFromTempDir(slowLogData->clusterId); monitorSendAllSlowLogFromTempDir(slowLogData->clusterId);
} }
@ -850,7 +836,7 @@ int32_t monitorPutData2MonitorQueue(MonitorSlowLogData data){
return -1; return -1;
} }
*slowLogData = data; *slowLogData = data;
tscDebug("[monitor] write slow log to queue, clusterId:%"PRIx64 " type:%d", slowLogData->clusterId, slowLogData->type); tscDebug("[monitor] write slow log to queue, clusterId:%"PRIx64 " type:%s, data:%s", slowLogData->clusterId, queueTypeStr[slowLogData->type], slowLogData->data);
if (taosWriteQitem(monitorQueue, slowLogData) == 0){ if (taosWriteQitem(monitorQueue, slowLogData) == 0){
tsem2_post(&monitorSem); tsem2_post(&monitorSem);
}else{ }else{