Merge pull request #15238 from taosdata/feature/3.0_mhli

fix(sync): raft store persist

source/libs/sync/src/syncAppendEntries.c

@@ -473,13 +473,13 @@ static bool syncNodeOnAppendEntriesBatchLogOK(SSyncNode* pSyncNode, SyncAppendEn
 
   SyncIndex myLastIndex = syncNodeGetLastIndex(pSyncNode);
   if (pMsg->prevLogIndex > myLastIndex) {
-    sDebug("vgId:%d sync log not ok, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
+    sDebug("vgId:%d, sync log not ok, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
     return false;
   }
 
   SyncTerm myPreLogTerm = syncNodeGetPreTerm(pSyncNode, pMsg->prevLogIndex + 1);
   if (myPreLogTerm == SYNC_TERM_INVALID) {
-    sDebug("vgId:%d sync log not ok2, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
+    sDebug("vgId:%d, sync log not ok2, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
     return false;
   }
 
@@ -487,7 +487,7 @@ static bool syncNodeOnAppendEntriesBatchLogOK(SSyncNode* pSyncNode, SyncAppendEn
     return true;
   }
 
-  sDebug("vgId:%d sync log not ok3, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
+  sDebug("vgId:%d, sync log not ok3, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
   return false;
 }
 
@@ -500,13 +500,13 @@ static bool syncNodeOnAppendEntriesLogOK(SSyncNode* pSyncNode, SyncAppendEntries
 
   SyncIndex myLastIndex = syncNodeGetLastIndex(pSyncNode);
   if (pMsg->prevLogIndex > myLastIndex) {
-    sDebug("vgId:%d sync log not ok, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
+    sDebug("vgId:%d, sync log not ok, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
     return false;
   }
 
   SyncTerm myPreLogTerm = syncNodeGetPreTerm(pSyncNode, pMsg->prevLogIndex + 1);
   if (myPreLogTerm == SYNC_TERM_INVALID) {
-    sDebug("vgId:%d sync log not ok2, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
+    sDebug("vgId:%d, sync log not ok2, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
     return false;
   }
 
@@ -514,7 +514,7 @@ static bool syncNodeOnAppendEntriesLogOK(SSyncNode* pSyncNode, SyncAppendEntries
     return true;
   }
 
-  sDebug("vgId:%d sync log not ok3, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
+  sDebug("vgId:%d, sync log not ok3, preindex:%" PRId64, pSyncNode->vgId, pMsg->prevLogIndex);
   return false;
 }
 

source/libs/sync/src/syncRaftStore.c

@@ -108,10 +108,10 @@ int32_t raftStoreSerialize(SRaftStore *pRaftStore, char *buf, size_t len) {
   cJSON *pRoot = cJSON_CreateObject();
 
   char u64Buf[128] = {0};
-  snprintf(u64Buf, sizeof(u64Buf), "" PRIu64 "", pRaftStore->currentTerm);
+  snprintf(u64Buf, sizeof(u64Buf), "%" PRIu64 "", pRaftStore->currentTerm);
   cJSON_AddStringToObject(pRoot, "current_term", u64Buf);
 
-  snprintf(u64Buf, sizeof(u64Buf), "" PRIu64 "", pRaftStore->voteFor.addr);
+  snprintf(u64Buf, sizeof(u64Buf), "%" PRIu64 "", pRaftStore->voteFor.addr);
   cJSON_AddStringToObject(pRoot, "vote_for_addr", u64Buf);
 
   cJSON_AddNumberToObject(pRoot, "vote_for_vgid", pRaftStore->voteFor.vgId);
@@ -142,11 +142,11 @@ int32_t raftStoreDeserialize(SRaftStore *pRaftStore, char *buf, size_t len) {
 
   cJSON *pCurrentTerm = cJSON_GetObjectItem(pRoot, "current_term");
   ASSERT(cJSON_IsString(pCurrentTerm));
-  sscanf(pCurrentTerm->valuestring, "" PRIu64 "", &(pRaftStore->currentTerm));
+  sscanf(pCurrentTerm->valuestring, "%" PRIu64 "", &(pRaftStore->currentTerm));
 
   cJSON *pVoteForAddr = cJSON_GetObjectItem(pRoot, "vote_for_addr");
   ASSERT(cJSON_IsString(pVoteForAddr));
-  sscanf(pVoteForAddr->valuestring, "" PRIu64 "", &(pRaftStore->voteFor.addr));
+  sscanf(pVoteForAddr->valuestring, "%" PRIu64 "", &(pRaftStore->voteFor.addr));
 
   cJSON *pVoteForVgid = cJSON_GetObjectItem(pRoot, "vote_for_vgid");
   pRaftStore->voteFor.vgId = pVoteForVgid->valueint;
@@ -188,11 +188,11 @@ cJSON *raftStore2Json(SRaftStore *pRaftStore) {
   cJSON *pRoot = cJSON_CreateObject();
 
   if (pRaftStore != NULL) {
-    snprintf(u64buf, sizeof(u64buf), "" PRIu64 "", pRaftStore->currentTerm);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pRaftStore->currentTerm);
     cJSON_AddStringToObject(pRoot, "currentTerm", u64buf);
 
     cJSON *pVoteFor = cJSON_CreateObject();
-    snprintf(u64buf, sizeof(u64buf), "" PRIu64 "", pRaftStore->voteFor.addr);
+    snprintf(u64buf, sizeof(u64buf), "%" PRIu64 "", pRaftStore->voteFor.addr);
     cJSON_AddStringToObject(pVoteFor, "addr", u64buf);
     {
       uint64_t u64 = pRaftStore->voteFor.addr;

source/libs/sync/test/sh/a.sh

+#!/bin/bash
+
+if [ $# != 1 ] ; then
+	echo "Uasge: $0 log-path"
+	echo ""
+	exit 1
+fi
+
+logpath=$1
+echo "logpath: ${logpath}"
+
+echo ""
+echo "clean old log ..."
+rm -f ${logpath}/log.*
+
+echo ""
+echo "generate log.dnode ..."
+for dnode in `ls ${logpath} | grep dnode`;do
+	echo "generate log.${dnode}"
+	cat ${logpath}/${dnode}/log/taosdlog.* | grep SYN > ${logpath}/log.${dnode}
+done
+
+echo ""
+echo "generate vgId ..."
+cat ${logpath}/log.dnode* | grep "vgId:" | grep -v ERROR | awk '{print $5}' | sort | uniq > ${logpath}/log.vgIds.tmp
+echo "all vgIds:" > ${logpath}/log.vgIds
+cat ${logpath}/log.dnode* | grep "vgId:" | grep -v ERROR | awk '{print $5}' | sort | uniq >> ${logpath}/log.vgIds
+for dnode in `ls ${logpath} | grep dnode | grep -v log`;do
+	echo "" >> ${logpath}/log.vgIds
+	echo "" >> ${logpath}/log.vgIds
+	echo "${dnode}:" >> ${logpath}/log.vgIds
+	cat ${logpath}/${dnode}/log/taosdlog.* | grep SYN | grep "vgId:" | grep -v ERROR | awk '{print $5}' | sort | uniq >> ${logpath}/log.vgIds 
+done
+
+echo ""
+echo "generate log.dnode.vgId ..."
+for logdnode in `ls ${logpath}/log.dnode*`;do
+	for vgId in `cat ${logpath}/log.vgIds.tmp`;do
+		rowNum=`cat ${logdnode} | grep "${vgId}" | awk 'BEGIN{rowNum=0}{rowNum++}END{print rowNum}'`
+		#echo "-----${rowNum}"
+		if [ $rowNum -gt 0 ] ; then
+			echo "generate ${logdnode}.${vgId}"
+			cat ${logdnode} | grep "${vgId}" > ${logdnode}.${vgId}
+		fi
+	done
+done
+
+echo ""
+echo "generate log.dnode.main ..."
+for file in `ls ${logpath}/log.dnode* | grep -v vgId`;do
+	echo "generate ${file}.main"
+	cat ${file} | awk '{ if(index($0, "sync open") > 0 || index($0, "sync close") > 0 || index($0, "become leader") > 0) {print $0} }' > ${file}.main
+done
+
+
+
+
+exit 0