fix(meta): check and rename old raft dir if exist. #1000187608

Signed-off-by: Victor1319 <zengxuewei@oppo.com>

metanode/manager.go

@@ -686,13 +686,14 @@ func (m *metadataManager) loadPartitions() (err error) {
 	}
 	syslog.Println("Start loadPartitions!!!")
 	var wg sync.WaitGroup
+	curTime := "_" + time.Now().Format(StaleMetadataTimeFormat)
 	for _, fileInfo := range fileInfoList {
 		if fileInfo.IsDir() && strings.HasPrefix(fileInfo.Name(), partitionPrefix) {
 			if isExpiredPartition(fileInfo.Name(), metaNodeInfo.PersistenceMetaPartitions) {
 				log.LogErrorf("loadPartitions: find expired partition[%s], rename it and you can delete it manually",
 					fileInfo.Name())
 				oldName := path.Join(m.rootDir, fileInfo.Name())
-				newName := path.Join(m.rootDir, ExpiredPartitionPrefix+fileInfo.Name())
+				newName := path.Join(m.rootDir, ExpiredPartitionPrefix+fileInfo.Name()+curTime)
 				os.Rename(oldName, newName)
 				continue
 			}

metanode/partition.go

@@ -870,7 +870,7 @@ func (mp *metaPartition) onStart(isCreate bool) (err error) {
 
 	log.LogWarnf("[before raft] get mp[%v] applied(%d),inodeCount(%d),dentryCount(%d)", mp.config.PartitionId, mp.applyID, mp.inodeTree.Len(), mp.dentryTree.Len())
 
-	if err = mp.startRaft(); err != nil {
+	if err = mp.startRaft(isCreate); err != nil {
 		err = errors.NewErrorf("[onStart] start raft id=%d: %s",
 			mp.config.PartitionId, err.Error())
 		return
@@ -900,7 +900,7 @@ func (mp *metaPartition) onStop() {
 	}
 }
 
-func (mp *metaPartition) startRaft() (err error) {
+func (mp *metaPartition) startRaft(isCreate bool) (err error) {
 	var (
 		heartbeatPort int
 		replicaPort   int
@@ -934,10 +934,11 @@ func (mp *metaPartition) startRaft() (err error) {
 	log.LogInfof("start partition id=%d,applyID:%v raft peers: %s",
 		mp.config.PartitionId, mp.applyID, peers)
 	pc := &raftstore.PartitionConfig{
-		ID:      mp.config.PartitionId,
+		ID:       mp.config.PartitionId,
-		Applied: mp.applyID,
+		Applied:  mp.applyID,
-		Peers:   peers,
+		Peers:    peers,
-		SM:      mp,
+		SM:       mp,
+		IsCreate: isCreate,
 	}
 	mp.raftPartition, err = mp.config.RaftStore.CreatePartition(pc)
 	if err == nil {

raftstore/config.go

@@ -64,13 +64,14 @@ type PeerAddress struct {
 
 // PartitionConfig defines the configuration properties for the partitions.
 type PartitionConfig struct {
-	ID      uint64
+	ID       uint64
-	Applied uint64
+	Applied  uint64
-	Leader  uint64
+	Leader   uint64
-	Term    uint64
+	Term     uint64
-	Peers   []PeerAddress
+	Peers    []PeerAddress
-	SM      PartitionFsm
+	SM       PartitionFsm
-	WalPath string
+	WalPath  string
+	IsCreate bool
 }
 
 func (p PeerAddress) String() string {

raftstore/raftstore.go

@@ -164,6 +164,28 @@ func (s *raftStore) CreatePartition(cfg *PartitionConfig) (p Partition, err erro
 		walPath = path.Join(cfg.WalPath, "wal_"+strconv.FormatUint(cfg.ID, 10))
 	}
 
+	if cfg.IsCreate {
+		logger.Warn("action[raftstore:CreatePartition] check old dir exist, raft[%v], wal %v",
+			cfg.ID, walPath)
+		if _, err = os.Stat(walPath); err != nil {
+			if !os.IsNotExist(err) {
+				logger.Warn("action[raftstore:CreatePartition] raft[%v], wal %v, err %s", cfg.ID, walPath, err.Error())
+				return nil, err
+			}
+		} else {
+			curTime := time.Now().Format("20060102150405.000000000")
+			backDirName := walPath + "_" + curTime + "_old"
+			if err = os.Rename(walPath, backDirName); err != nil {
+				logger.Warn("action[raftstore:CreatePartition] rename old failed, raft[%v], wal %v, old %v, err %s",
+					cfg.ID, walPath, backDirName, err.Error())
+				return nil, err
+			}
+
+			logger.Warn("action[raftstore:CreatePartition] rename old succ, raft[%v], wal %v, old %v",
+				cfg.ID, walPath, backDirName)
+		}
+	}
+
 	wc := &wal.Config{}
 	ws, err := wal.NewStorage(walPath, wc)
 	if err != nil {
@@ -179,7 +201,8 @@ func (s *raftStore) CreatePartition(cfg *PartitionConfig) (p Partition, err erro
 			peerAddress.ReplicaPort,
 		)
 	}
-	logger.Info("action[raftstore:CreatePartition] raft config applied [%v] id:%d", cfg.Applied, cfg.ID)
+	logger.Info("action[raftstore:CreatePartition] raft config applied [%v] id:%d, isCreate %v",
+		cfg.Applied, cfg.ID, cfg.IsCreate)
 	rc := &raft.RaftConfig{
 		ID:           cfg.ID,
 		Peers:        peers,