fix(meta): increase the timeout when fetch info from master in register and mp start procedure

close:#22677404
Signed-off-by: true1064 <tangjingyu@oppo.com>

metanode/inode.go

@@ -995,7 +995,7 @@ func (i *Inode) UnmarshalInodeValue(buff *bytes.Buffer) (err error) {
 		return
 	}
 	if err = binary.Read(buff, binary.BigEndian, &i.Uid); err != nil {
-		err = UnmarshalInodeFiledError("Type", err)
+		err = UnmarshalInodeFiledError("Uid", err)
 		return
 	}
 	if err = binary.Read(buff, binary.BigEndian, &i.Gid); err != nil {

metanode/metanode.go

@@ -37,7 +37,7 @@ import (
 )
 
 var (
-	clusterInfo *proto.ClusterInfo
+	gClusterInfo *proto.ClusterInfo
 	// masterClient   *masterSDK.MasterClient
 	masterClient              *masterSDK.MasterCLientWithResolver
 	configTotalMem            uint64
@@ -471,60 +471,59 @@ func (m *MetaNode) stopMetaManager() {
 }
 
 func (m *MetaNode) register() (err error) {
-	step := 0
+	tryCnt := 0
 	var nodeAddress string
 	var volsForbidWriteOpVerMsg string
 	var nodeForbidWriteOpOfProtoVerMsg string
 
 	for {
-		if step < 1 {
-			clusterInfo, err = getClusterInfo()
-			if err != nil {
-				log.LogErrorf("[register] %s", err.Error())
+		tryCnt++
+		gClusterInfo, err = getClusterInfo()
+		if err != nil {
+			log.LogErrorf("[register] tryCnt(%v), getClusterInfo err: %s", tryCnt, err.Error())
+			time.Sleep(3 * time.Second)
+			continue
+		}
+		if m.localAddr == "" {
+			m.localAddr = gClusterInfo.Ip
+		}
+		m.clusterUuid = gClusterInfo.ClusterUuid
+		m.clusterUuidEnable = gClusterInfo.ClusterUuidEnable
+		m.clusterEnableSnapshot = gClusterInfo.ClusterEnableSnapshot
+		clusterEnableSnapshot = m.clusterEnableSnapshot
+		m.clusterId = gClusterInfo.Cluster
+		nodeAddress = m.localAddr + ":" + m.listen
+		m.nodeForbidWriteOpOfProtoVer0 = gClusterInfo.ForbidWriteOpOfProtoVer0
+		nodeForbidWriteOpOfProtoVerMsg = fmt.Sprintf("[register] from master, node forbid write Operate Of proto version-0: %v",
+			m.nodeForbidWriteOpOfProtoVer0)
+
+		volListForbidWriteOpOfProtoVer0 := make([]string, 0)
+		var volListForbidFromMaster *proto.VolListForbidWriteOpOfProtoVer0
+		if volListForbidFromMaster, err = getVolListForbiddenWriteOpOfProtoVer0(); err != nil {
+			if strings.Contains(err.Error(), proto.KeyWordInHttpApiNotSupportErr) {
+				// master may be lower version and has no this API
+				volsForbidWriteOpVerMsg = fmt.Sprintf("[register] master version has no api GetVolListForbiddenWriteOpOfProtoVer0, ues default value(false)")
+			} else {
+				log.LogErrorf("[register] tryCnt(%v), failed to get volume list forbidden write op of proto version-0 from master, err: %v", tryCnt, err)
+				time.Sleep(3 * time.Second)
 				continue
 			}
-			if m.localAddr == "" {
-				m.localAddr = clusterInfo.Ip
-			}
-			m.clusterUuid = clusterInfo.ClusterUuid
-			m.clusterUuidEnable = clusterInfo.ClusterUuidEnable
-			m.clusterEnableSnapshot = clusterInfo.ClusterEnableSnapshot
-			clusterEnableSnapshot = m.clusterEnableSnapshot
-			m.clusterId = clusterInfo.Cluster
-			nodeAddress = m.localAddr + ":" + m.listen
-			m.nodeForbidWriteOpOfProtoVer0 = clusterInfo.ForbidWriteOpOfProtoVer0
-			nodeForbidWriteOpOfProtoVerMsg = fmt.Sprintf("[register] from master, node forbid write Operate Of proto version-0: %v",
-				m.nodeForbidWriteOpOfProtoVer0)
-
-			volListForbidWriteOpOfProtoVer0 := make([]string, 0)
-			var volListForbidFromMaster *proto.VolListForbidWriteOpOfProtoVer0
-			if volListForbidFromMaster, err = getVolListForbiddenWriteOpOfProtoVer0(); err != nil {
-				if strings.Contains(err.Error(), proto.KeyWordInHttpApiNotSupportErr) {
-					// master may be lower version and has no this API
-					volsForbidWriteOpVerMsg = fmt.Sprintf("[registerToMaster] master version has no api GetVolListForbiddenWriteOpOfProtoVer0, ues default value(false)")
-				} else {
-					log.LogErrorf("[registerToMaster] failed to get volume list forbidden write op of proto version-0 from master, err: %v", err)
-					continue
-				}
-			} else {
-				volListForbidWriteOpOfProtoVer0 = volListForbidFromMaster.VolsForbidWriteOpOfProtoVer0
-				volsForbidWriteOpVerMsg = fmt.Sprintf("[registerToMaster] from master, volumes forbid write operate of proto version-0: %v",
-					volListForbidWriteOpOfProtoVer0)
-			}
-			volMapForbidWriteOpOfProtoVer0 := make(map[string]struct{})
-			for _, vol := range volListForbidWriteOpOfProtoVer0 {
-				if _, ok := volMapForbidWriteOpOfProtoVer0[vol]; !ok {
-					volMapForbidWriteOpOfProtoVer0[vol] = struct{}{}
-				}
-			}
-			m.VolsForbidWriteOpOfProtoVer0 = volMapForbidWriteOpOfProtoVer0
-
-			step++
+		} else {
+			volListForbidWriteOpOfProtoVer0 = volListForbidFromMaster.VolsForbidWriteOpOfProtoVer0
+			volsForbidWriteOpVerMsg = fmt.Sprintf("[register] from master, volumes forbid write operate of proto version-0: %v",
+				volListForbidWriteOpOfProtoVer0)
 		}
+		volMapForbidWriteOpOfProtoVer0 := make(map[string]struct{})
+		for _, vol := range volListForbidWriteOpOfProtoVer0 {
+			if _, ok := volMapForbidWriteOpOfProtoVer0[vol]; !ok {
+				volMapForbidWriteOpOfProtoVer0[vol] = struct{}{}
+			}
+		}
+		m.VolsForbidWriteOpOfProtoVer0 = volMapForbidWriteOpOfProtoVer0
 
 		var nodeID uint64
 		if nodeID, err = masterClient.NodeAPI().AddMetaNodeWithAuthNode(nodeAddress, m.zoneName, m.serviceIDKey); err != nil {
-			log.LogErrorf("register: register to master fail: address(%v) err(%s)", nodeAddress, err)
+			log.LogErrorf("[register] tryCnt(%v), register to master fail: address(%v) err(%s)", tryCnt, nodeAddress, err)
 			time.Sleep(3 * time.Second)
 			continue
 		}

metanode/nodeinfo.go

@@ -69,7 +69,6 @@ func (m *MetaNode) stopUpdateNodeInfo() {
 }
 
 func (m *MetaNode) updateNodeInfo() {
-	// clusterInfo, err := getClusterInfo()
 	clusterInfo, err := masterClient.AdminAPI().GetClusterInfo()
 	if err != nil {
 		log.LogErrorf("[updateNodeInfo] %s", err.Error())
@@ -86,6 +85,4 @@ func (m *MetaNode) updateNodeInfo() {
 		atomic.StoreUint32(&dirChildrenNumLimit, clusterInfo.DirChildrenNumLimit)
 		log.LogInfof("updateNodeInfo: DirChildrenNumLimit(%v)", clusterInfo.DirChildrenNumLimit)
 	}
-
-	// updateDirChildrenNumLimit(clusterInfo.DirChildrenNumLimit)
 }

metanode/partition.go

@@ -537,7 +537,7 @@ func (ew *BlobStoreClientWrapper) getBlobStoreClient() (blobClient *blobstore.Bl
 		return nil, create, err
 	}
 
-	log.LogDebugf("[getBlobStoreClient] addr(%v) create blobstore client success", clusterInfo.EbsAddr)
+	log.LogDebugf("[getBlobStoreClient] addr(%v) create blobstore client success", gClusterInfo.EbsAddr)
 	ew.blobClient = blobClient
 	ew.lastTryCreateTime = 0
 	create = true
@@ -849,15 +849,19 @@ func (mp *metaPartition) onStart(isCreate bool) (err error) {
 		return
 	}
 
-	// set EBS Client
-	if clusterInfo, err = masterClient.AdminAPI().GetClusterInfo(); err != nil {
-		log.LogErrorf("action[onStart] GetClusterInfo err[%v]", err)
-		return
-	}
-
 	var volumeInfo *proto.SimpleVolView
-	if volumeInfo, err = masterClient.AdminAPI().GetVolumeSimpleInfo(mp.config.VolName); err != nil {
-		log.LogErrorf("action[onStart] GetVolumeSimpleInfo err[%v]", err)
+	retryCnt := 0
+	for ; retryCnt < 200; retryCnt++ {
+		if volumeInfo, err = masterClient.AdminAPI().GetVolumeSimpleInfo(mp.config.VolName); err != nil {
+			log.LogWarnf("[onStart] vol(%v) mpId(%d) retryCnt(%v), GetVolumeSimpleInfo err[%v]",
+				mp.config.VolName, mp.config.PartitionId, retryCnt, err)
+			time.Sleep(3 * time.Second)
+			continue
+		}
+	}
+	if err != nil {
+		log.LogErrorf("[onStart] vol(%v) mpId(%d), after retryCnt(%v) failed to GetVolumeSimpleInfo: %v",
+			mp.config.VolName, mp.config.PartitionId, retryCnt, err)
 		return
 	}
 
@@ -867,13 +871,32 @@ func (mp *metaPartition) onStart(isCreate bool) (err error) {
 	}
 
 	mp.volType = volumeInfo.VolType
-	mp.volStorageClass = volumeInfo.VolStorageClass // TODO: tangjingyu handle compatibility with old version master
+	if proto.IsValidStorageClass(volumeInfo.VolStorageClass) {
+		mp.volStorageClass = volumeInfo.VolStorageClass
+		log.LogInfof("[onStart] vol(%v) mpId(%v), from master VolStorageClass(%v)",
+			mp.config.VolName, mp.config.PartitionId, proto.StorageClassString(mp.volStorageClass))
+	} else if volumeInfo.VolStorageClass == proto.StorageClass_Unspecified {
+		// handle compatibility with old version master which has no field VolStorageClass
+		if proto.IsValidStorageClass(legacyReplicaStorageClass) {
+			mp.volStorageClass = legacyReplicaStorageClass
+			log.LogWarnf("[onStart] vol(%v) mpId(%v), use conf legacyReplicaStorageClass(%v)",
+				mp.config.VolName, mp.config.PartitionId, proto.StorageClassString(legacyReplicaStorageClass))
+		} else {
+			err = errors.NewErrorf("[onStart] vol(%v) mpId(%d), master invalid volStorageClass(%v) and conf legacyReplicaStorageClass not set",
+				mp.config.VolName, mp.config.PartitionId, volumeInfo.VolStorageClass)
+			return
+		}
+	} else {
+		err = errors.NewErrorf("[onStart] vol(%v) mpId(%d), get from master invalid volStorageClass(%v)",
+			mp.config.VolName, mp.config.PartitionId, volumeInfo.VolStorageClass)
+		return
+	}
 
 	if proto.IsCold(mp.volType) || proto.IsVolSupportStorageClass(volumeInfo.AllowedStorageClass, proto.StorageClass_BlobStore) {
 		mp.blobClientWrapper, err = NewBlobStoreClientWrapper(access.Config{
 			ConnMode: access.NoLimitConnMode,
 			Consul: access.ConsulConfig{
-				Address: clusterInfo.EbsAddr,
+				Address: gClusterInfo.EbsAddr, // gClusterInfo is fetched from master in register procedure
 			},
 			MaxSizePutOnce: int64(volumeInfo.ObjBlockSize),
 			Logger:         &access.Logger{Filename: path.Join(log.LogDir, "ebs.log")},
@@ -881,11 +904,11 @@ func (mp *metaPartition) onStart(isCreate bool) (err error) {
 
 		if err != nil {
 			log.LogWarnf("action[onStart] mp(%v) blobStoreAddr(%v), create blobstore client err[%v], but still start mp and will try create blobstore later",
-				mp.config.PartitionId, clusterInfo.EbsAddr, err)
+				mp.config.PartitionId, gClusterInfo.EbsAddr, err)
 			// not return err here, blobstore client may be created latter
 		} else {
 			log.LogInfof("action[onStart] mp(%v) blobStoreAddr(%v), create blobstore client success",
-				mp.config.PartitionId, clusterInfo.EbsAddr)
+				mp.config.PartitionId, gClusterInfo.EbsAddr)
 		}
 	}
 

metanode/partition_fsm.go

@@ -1001,7 +1001,7 @@ func (mp *metaPartition) HandleLeaderChange(leader uint64) {
 			exporter.Warning(fmt.Sprintf("[HandleLeaderChange] pid %v init root inode id: %s.", mp.config.PartitionId, err.Error()))
 		}
 		ino := NewInode(id, proto.Mode(os.ModePerm|os.ModeDir))
-		ino.StorageClass = mp.volStorageClass // TODO: tangjingyu handle compatibility with old version master
+		ino.StorageClass = mp.volStorageClass
 		go mp.initInode(ino)
 	}
 	// refresh forbidden migration list