gian/cubefs
1
0
Fork 0
mirror of https://github.com/cubefs/cubefs.git synced 2026-08-02 02:00:56 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

feat(masterdatameta): support volume to forbidden write operate codes of lower packet protocol version

Browse Source 
close#22531837

Signed-off-by: true1064 <tangjingyu@oppo.com>
This commit is contained in:
true1064 2024-09-19 20:53:45 +08:00 committed by AmazingChi
parent 9d23e1660f
commit 9f7ffeb194
38 changed files with 577 additions and 346 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
cli/cmd/cluster.go
View File

@ -360,7 +360,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
if forbidWriteOpOfProtoVersion0 != "" {
if _, err = strconv.ParseBool(forbidWriteOpOfProtoVersion0); err != nil {
err = fmt.Errorf("forbidWriteOpOfProtoVersion0(%v) should be true or false", forbidWriteOpOfProtoVersion0)
err = fmt.Errorf("param forbidWriteOpOfProtoVersion0(%v) should be true or false", forbidWriteOpOfProtoVersion0)
return
}
}
@ -398,7 +398,7 @@ func newClusterSetParasCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().StringVar(&decommissionDpLimit, CliFlagDecommissionDpLimit, "", "Limit for parallel decommission dp")
cmd.Flags().StringVar(&decommissionDiskLimit, CliFlagDecommissionDiskLimit, "", "Limit for parallel decommission disk")
cmd.Flags().StringVar(&forbidWriteOpOfProtoVersion0, CliForbidWriteOpOfProtoVersion0, "",
"set datanode and metanode whether forbid write operate of packet protocol version-0: [true | false]")
"set datanode and metanode whether forbid write operate of packet whose protocol version is version-0: [true | false]")
return cmd
}

6
cli/cmd/compatibility.go
View File

@ -78,9 +78,9 @@ func newMetaCompatibilityCmd() *cobra.Command {
PartitionId: id,
}
mp := metanode.NewMetaPartition(mpcfg, nil)
if mp == nil {
return
}
// if mp == nil {
// return
// }
err = mp.LoadSnapshot(snapshotPath)
if err != nil {
return

2
cli/cmd/const.go
View File

@ -151,7 +151,7 @@ const (
CliFlagAllowedStorageClass = "allowedStorageClass"
CliFlagVolStorageClass = "volStorageClass"
CliFlagMediaType = "mediaType"
CliForbidWriteOpOfProtoVersion0 = "forbiddenWriteOpOfProtoVersion0"
CliForbidWriteOpOfProtoVersion0 = "forbidWriteOpOfProtoVersion0"
// CliFlagSetDataPartitionCount = "count" use dp-count instead
// Shorthand format of resource name

1
cli/cmd/fmt.go
View File

@ -463,6 +463,7 @@ func formatDataPartitionInfo(partition *proto.DataPartitionInfo) string {
sb.WriteString(fmt.Sprintf("Forbidden : %v\n", partition.Forbidden))
sb.WriteString(fmt.Sprintf("MediaType : %v\n", proto.MediaTypeString(partition.MediaType)))
sb.WriteString(fmt.Sprintf("ForbidWriteOpOfProtoVer0 : %v\n", partition.ForbidWriteOpOfProtoVer0))
sb.WriteString("\n")
sb.WriteString("Replicas : \n")
sb.WriteString(fmt.Sprintf("%v\n", formatDataReplicaTableHeader()))
for idx, replica := range partition.Replicas {

20
cli/cmd/vol.go
View File

@ -316,6 +316,7 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
var optAccessTimeValidInterval int64
var optEnablePersistAccessTime string
var optVolStorageClass int
var optForbidWriteOpOfProtoVer0 string
confirmString := strings.Builder{}
var vv *proto.SimpleVolView
@ -695,6 +696,23 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
proto.StorageClassString(vv.VolStorageClass)))
}
if optForbidWriteOpOfProtoVer0 != "" {
enable := false
if enable, err = strconv.ParseBool(optForbidWriteOpOfProtoVer0); err != nil {
err = fmt.Errorf("param forbidWriteOpOfProtoVersion0(%v) should be true or false", optForbidWriteOpOfProtoVer0)
return
}
if vv.ForbidWriteOpOfProtoVer0 != enable {
isChange = true
confirmString.WriteString(fmt.Sprintf(" ForbidWriteOpOfProtoVer0 : %v -> %v\n", vv.ForbidWriteOpOfProtoVer0, enable))
vv.ForbidWriteOpOfProtoVer0 = enable
} else {
confirmString.WriteString(fmt.Sprintf(" ForbidWriteOpOfProtoVer0 : %v\n", vv.ForbidWriteOpOfProtoVer0))
}
} else {
confirmString.WriteString(fmt.Sprintf(" ForbidWriteOpOfProtoVer0 : %v\n", vv.ForbidWriteOpOfProtoVer0))
}
if err != nil {
return
}
@ -759,6 +777,8 @@ func newVolUpdateCmd(client *master.MasterClient) *cobra.Command {
cmd.Flags().Int64Var(&optTrashInterval, CliFlagTrashInterval, -1, "The retention period for files in trash")
cmd.Flags().Int64Var(&optAccessTimeValidInterval, CliFlagAccessTimeValidInterval, -1, "Effective time interval for accesstime, at least 1800 [Unit: second]")
cmd.Flags().StringVar(&optEnablePersistAccessTime, CliFlagEnablePersistAccessTime, "", "true/false to enable/disable persisting access time")
cmd.Flags().StringVar(&optForbidWriteOpOfProtoVer0, CliForbidWriteOpOfProtoVersion0, "",
"set volume forbid write operates of packet whose protocol version is version-0: [true | false]")
return cmd
}

4
client/fs/file.go
View File

@ -420,8 +420,8 @@ func (f *File) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.Wri
ino := f.info.Inode
reqlen := len(req.Data)
log.LogDebugf("TRACE Write enter: ino(%v) offset(%v) len(%v) flags(%v) fileflags(%v) quotaIds(%v) req(%v)",
ino, req.Offset, reqlen, req.Flags, req.FileFlags, f.info.QuotaInfos, req)
log.LogDebugf("TRACE Write enter: ino(%v) storageClass(%v) offset(%v) len(%v) flags(%v) fileflags(%v) quotaIds(%v) req(%v)",
ino, f.info.StorageClass, req.Offset, reqlen, req.Flags, req.FileFlags, f.info.QuotaInfos, req)
if proto.IsHot(f.super.volType) || proto.IsStorageClassReplica(f.info.StorageClass) {
filesize, _ := f.fileSize(ino)
if req.Offset > int64(filesize) && reqlen == 1 && req.Data[0] == 0 {

18
datanode/partition.go
View File

@ -166,6 +166,14 @@ func (dp *DataPartition) SetForbidden(status bool) {
dp.config.Forbidden = status
}
func (dp *DataPartition) IsForbidWriteOpOfProtoVer0() bool {
return dp.config.ForbidWriteOpOfProtoVer0
}
func (dp *DataPartition) SetForbidWriteOpOfProtoVer0(status bool) {
dp.config.ForbidWriteOpOfProtoVer0 = status
}
func (dp *DataPartition) GetRepairBlockSize() (size uint64) {
size = dp.config.DpRepairBlockSize
if size == 0 {
@ -389,6 +397,16 @@ func newDataPartition(dpCfg *dataPartitionCfg, disk *Disk, isCreate bool) (dp *D
}
atomic.StoreUint64(&partition.recoverErrCnt, 0)
log.LogInfof("action[newDataPartition] dp %v replica num %v", partitionID, dpCfg.ReplicaNum)
VolsForbidWriteOpOfProtoVer0 := disk.dataNode.VolsForbidWriteOpOfProtoVer0
if _, ok := VolsForbidWriteOpOfProtoVer0[partition.volumeID]; ok {
partition.SetForbidWriteOpOfProtoVer0(true)
} else {
partition.SetForbidWriteOpOfProtoVer0(false)
}
log.LogInfof("[newDataPartition] vol(%v) dp(%v) IsForbidWriteOpOfProtoVer0: %v",
dpCfg.VolName, partitionID, partition.IsForbidWriteOpOfProtoVer0())
partition.replicasInit()
partition.extentStore, err = storage.NewExtentStore(partition.path, dpCfg.PartitionID, dpCfg.PartitionSize,
partition.partitionType, isCreate)

31
datanode/partition_raft.go
View File

@ -38,21 +38,22 @@ import (
)
type dataPartitionCfg struct {
VolName string `json:"vol_name"`
ClusterID string `json:"cluster_id"`
PartitionID uint64 `json:"partition_id"`
PartitionSize int `json:"partition_size"`
PartitionType int `json:"partition_type"`
Peers []proto.Peer `json:"peers"`
Hosts []string `json:"hosts"`
NodeID uint64 `json:"-"`
RaftStore raftstore.RaftStore `json:"-"`
ReplicaNum int
VerSeq uint64 `json:"ver_seq"`
CreateType int
Forbidden bool
DpRepairBlockSize uint64
IsEnableSnapshot bool
VolName string `json:"vol_name"`
ClusterID string `json:"cluster_id"`
PartitionID uint64 `json:"partition_id"`
PartitionSize int `json:"partition_size"`
PartitionType int `json:"partition_type"`
Peers []proto.Peer `json:"peers"`
Hosts []string `json:"hosts"`
NodeID uint64 `json:"-"`
RaftStore raftstore.RaftStore `json:"-"`
ReplicaNum int
VerSeq uint64 `json:"ver_seq"`
CreateType int
Forbidden bool
DpRepairBlockSize uint64
IsEnableSnapshot bool
ForbidWriteOpOfProtoVer0 bool
}
func (dp *DataPartition) raftPort() (heartbeat, replica int, err error) {

27
datanode/server.go
View File

@ -205,8 +205,9 @@ type DataNode struct {
diskUnavailablePartitionErrorCount uint64 // disk status becomes unavailable when disk error partition count reaches this value
started int32
dpBackupTimeout time.Duration
mediaType uint32 // type of storage hardware media
forbidWriteOpOfProtoVer0 bool
mediaType uint32 // type of storage hardware medi
nodeForbidWriteOpOfProtoVer0 bool // whether forbid by node granularity,
VolsForbidWriteOpOfProtoVer0 map[string]struct{} // whether forbid by volume granularity,
}
type verOp2Phase struct {
@ -661,11 +662,23 @@ func (s *DataNode) register(cfg *config.Config) {
if LocalIP == "" {
LocalIP = string(ci.Ip)
}
s.forbidWriteOpOfProtoVer0 = ci.ForbidWriteOpOfProtoVer0
forbidWriteOpVerMsg := fmt.Sprintf("action[registerToMaster] from master forbidWriteOpOfProtoVer0: %v",
s.forbidWriteOpOfProtoVer0)
log.LogInfo(forbidWriteOpVerMsg)
syslog.Printf("%v \n", forbidWriteOpVerMsg)
s.nodeForbidWriteOpOfProtoVer0 = ci.ForbidWriteOpOfProtoVer0
nodeForbidWriteOpVerMsg := fmt.Sprintf("action[registerToMaster] from master, node forbid write Operate Of proto version-0: %v",
s.nodeForbidWriteOpOfProtoVer0)
log.LogInfo(nodeForbidWriteOpVerMsg)
syslog.Printf("%v\n", nodeForbidWriteOpVerMsg)
volsForbidWriteOpOfProtoVer0 := make(map[string]struct{})
for _, vol := range ci.VolsForbidWriteOpOfProtoVer0 {
if _, ok := volsForbidWriteOpOfProtoVer0[vol]; !ok {
volsForbidWriteOpOfProtoVer0[vol] = struct{}{}
}
}
s.VolsForbidWriteOpOfProtoVer0 = volsForbidWriteOpOfProtoVer0
volsForbidWriteOpVerMsg := fmt.Sprintf("action[registerToMaster] from master, volumes forbid write operate of proto version-0: %v",
ci.VolsForbidWriteOpOfProtoVer0)
log.LogInfo(volsForbidWriteOpVerMsg)
syslog.Printf("%v\n", volsForbidWriteOpVerMsg)
s.localServerAddr = fmt.Sprintf("%s:%v", LocalIP, s.port)
if !util.IsIPV4(LocalIP) {

58
datanode/space_manager.go
View File

@ -491,20 +491,21 @@ func (manager *SpaceManager) DetachDataPartition(partitionID uint64) {
func (manager *SpaceManager) CreatePartition(request *proto.CreateDataPartitionRequest) (dp *DataPartition, err error) {
dpCfg := &dataPartitionCfg{
PartitionID: request.PartitionId,
VolName: request.VolumeId,
Peers: request.Members,
Hosts: request.Hosts,
RaftStore: manager.raftStore,
NodeID: manager.nodeID,
ClusterID: manager.clusterID,
PartitionSize: request.PartitionSize,
PartitionType: int(request.PartitionTyp),
ReplicaNum: request.ReplicaNum,
VerSeq: request.VerSeq,
CreateType: request.CreateType,
Forbidden: false,
IsEnableSnapshot: manager.dataNode.clusterEnableSnapshot,
PartitionID: request.PartitionId,
VolName: request.VolumeId,
Peers: request.Members,
Hosts: request.Hosts,
RaftStore: manager.raftStore,
NodeID: manager.nodeID,
ClusterID: manager.clusterID,
PartitionSize: request.PartitionSize,
PartitionType: int(request.PartitionTyp),
ReplicaNum: request.ReplicaNum,
VerSeq: request.VerSeq,
CreateType: request.CreateType,
Forbidden: false,
IsEnableSnapshot: manager.dataNode.clusterEnableSnapshot,
ForbidWriteOpOfProtoVer0: false,
}
log.LogInfof("action[CreatePartition] dp %v dpCfg.Peers %v request.Members %v",
dpCfg.PartitionID, dpCfg.Peers, request.Members)
@ -558,7 +559,7 @@ func (manager *SpaceManager) DeletePartition(dpID uint64, force bool) (err error
}
func (s *DataNode) buildHeartBeatResponse(response *proto.DataNodeHeartbeatResponse,
volNames map[string]struct{}, dpRepairBlockSize map[string]uint64, reqID string,
forbiddenVols map[string]struct{}, dpRepairBlockSize map[string]uint64, reqID string,
) {
response.Status = proto.TaskSucceeds
stat := s.space.Stats()
@ -576,7 +577,7 @@ func (s *DataNode) buildHeartBeatResponse(response *proto.DataNodeHeartbeatRespo
stat.Unlock()
response.ZoneName = s.zoneName
response.ReceivedForbidWriteOpOfProtoVer0 = s.forbidWriteOpOfProtoVer0
response.ReceivedForbidWriteOpOfProtoVer0 = s.nodeForbidWriteOpOfProtoVer0
response.PartitionReports = make([]*proto.DataPartitionReport, 0)
space := s.space
begin := time.Now()
@ -599,20 +600,35 @@ func (s *DataNode) buildHeartBeatResponse(response *proto.DataNodeHeartbeatRespo
TriggerDiskError: atomic.LoadUint64(&partition.diskErrCnt) > 0,
}
log.LogDebugf("action[Heartbeats] dpid(%v), status(%v) total(%v) used(%v) leader(%v) isLeader(%v) "+
"TriggerDiskError(%v) reqId(%v) testID(%v)cost(%v).",
"TriggerDiskError(%v) reqId(%v) testID(%v) cost(%v).",
vr.PartitionID, vr.PartitionStatus, vr.Total, vr.Used, leaderAddr, vr.IsLeader, vr.TriggerDiskError,
reqID, testID, time.Now().Sub(begin2))
respLock.Lock()
response.PartitionReports = append(response.PartitionReports, vr)
respLock.Unlock()
begin2 = time.Now()
if len(volNames) != 0 {
if _, ok := volNames[partition.volumeID]; ok {
if len(forbiddenVols) != 0 {
if _, ok := forbiddenVols[partition.volumeID]; ok {
partition.SetForbidden(true)
} else {
partition.SetForbidden(false)
}
}
oldVal := partition.IsForbidWriteOpOfProtoVer0()
VolsForbidWriteOpOfProtoVer0 := s.VolsForbidWriteOpOfProtoVer0
if _, ok := VolsForbidWriteOpOfProtoVer0[partition.volumeID]; ok {
partition.SetForbidWriteOpOfProtoVer0(true)
} else {
partition.SetForbidWriteOpOfProtoVer0(false)
}
newVal := partition.IsForbidWriteOpOfProtoVer0()
if oldVal != newVal {
log.LogInfof("[Heartbeats] vol(%v) dpId(%v) IsForbidWriteOpOfProtoVer0 change to %v",
partition.volumeID, partition.partitionID, newVal)
}
size := uint64(proto.DefaultDpRepairBlockSize)
if len(dpRepairBlockSize) != 0 {
var ok bool
@ -620,8 +636,8 @@ func (s *DataNode) buildHeartBeatResponse(response *proto.DataNodeHeartbeatRespo
size = proto.DefaultDpRepairBlockSize
}
}
log.LogDebugf("action[Heartbeats] volume(%v) dp(%v) repair block size(%v) current size(%v) reqId(%v) testID(%v) cost(%v)",
partition.volumeID, partition.partitionID, size, partition.GetRepairBlockSize(), reqID, testID, time.Now().Sub(begin2))
log.LogDebugf("action[Heartbeats] volume(%v) dp(%v) repair block size(%v) current size(%v) reqId(%v) testID(%v) nodeForbidWriteOpOfProtoVer0(%v) cost(%v)",
partition.volumeID, partition.partitionID, size, partition.GetRepairBlockSize(), reqID, testID, partition.IsForbidWriteOpOfProtoVer0(), time.Now().Sub(begin2))
if partition.GetRepairBlockSize() != size {
partition.SetRepairBlockSize(size)
}

61
datanode/wrap_operator.go
View File

@ -84,14 +84,6 @@ func isColdVolExtentDelErr(p *repl.Packet) bool {
return false
}
func (s *DataNode) isWriteOpOfProtoVersionForbidden(pktProtoVersion uint32) (forbidden bool) {
if pktProtoVersion != proto.PacketProtoVersion0 {
return false
}
return s.forbidWriteOpOfProtoVer0
}
func (s *DataNode) OperatePacket(p *repl.Packet, c net.Conn) (err error) {
var (
tpLabels map[string]string
@ -245,8 +237,7 @@ func (s *DataNode) handlePacketToCreateExtent(p *repl.Packet) {
}()
partition := p.Object.(*DataPartition)
if s.isWriteOpOfProtoVersionForbidden(p.ProtoVersion) {
err = fmt.Errorf("%v %v", storage.WriteOpOfProtoVerForbidden, p.ProtoVersion)
if err = s.checkForbidWriteOpOfProtoVer0(p, partition); err != nil {
return
}
@ -571,10 +562,10 @@ func (s *DataNode) handleHeartbeatPacket(p *repl.Packet) {
s.dpBackupTimeout = dpBackupTimeout
log.LogDebugf("handleHeartbeatPacket receive req(%v) dpBackupTimeout(%v)", task.RequestID, dpBackupTimeout)
if s.forbidWriteOpOfProtoVer0 != request.NotifyForbidWriteOpOfProtoVer0 {
log.LogWarnf("[handleHeartbeatPacket] change forbidWriteOpOfProtoVer0, old(%v) new(%v)",
s.forbidWriteOpOfProtoVer0, request.NotifyForbidWriteOpOfProtoVer0)
s.forbidWriteOpOfProtoVer0 = request.NotifyForbidWriteOpOfProtoVer0
if s.nodeForbidWriteOpOfProtoVer0 != request.NotifyForbidWriteOpOfProtoVer0 {
log.LogWarnf("[handleHeartbeatPacket] change nodeForbidWriteOpOfProtoVer0, old(%v) new(%v)",
s.nodeForbidWriteOpOfProtoVer0, request.NotifyForbidWriteOpOfProtoVer0)
s.nodeForbidWriteOpOfProtoVer0 = request.NotifyForbidWriteOpOfProtoVer0
}
log.LogDebugf("handleHeartbeatPacket receive req(%v)", task.RequestID)
@ -589,6 +580,15 @@ func (s *DataNode) handleHeartbeatPacket(p *repl.Packet) {
forbiddenVols[vol] = struct{}{}
}
}
volsForbidWriteOpOfProtoVer0 := make(map[string]struct{})
for _, vol := range request.VolsForbidWriteOpOfProtoVer0 {
if _, ok := volsForbidWriteOpOfProtoVer0[vol]; !ok {
volsForbidWriteOpOfProtoVer0[vol] = struct{}{}
}
}
s.VolsForbidWriteOpOfProtoVer0 = volsForbidWriteOpOfProtoVer0
s.buildHeartBeatResponse(response, forbiddenVols, request.VolDpRepairBlockSize, task.RequestID)
log.LogDebugf("handleHeartbeatPacket buildHeartBeatResponse req(%v) cost %v",
task.RequestID, time.Now().Sub(begin))
@ -837,6 +837,29 @@ func (s *DataNode) handleBatchMarkDeletePacket(p *repl.Packet, c net.Conn) {
}
}
func (s *DataNode) checkForbidWriteOpOfProtoVer0(p *repl.Packet, dp *DataPartition) (err error) {
if p.Opcode == proto.OpBackupWrite {
// this opCode only used by fsck
return nil
}
if p.ProtoVersion != proto.PacketProtoVersion0 {
return nil
}
if s.nodeForbidWriteOpOfProtoVer0 {
err = fmt.Errorf("%v %v", storage.ClusterForbidWriteOpOfProtoVer, p.ProtoVersion)
return
}
if dp.IsForbidWriteOpOfProtoVer0() {
err = fmt.Errorf("%v %v", storage.VolForbidWriteOpOfProtoVer, p.ProtoVersion)
return
}
return nil
}
// Handle OpWrite packet.
func (s *DataNode) handleWritePacket(p *repl.Packet) {
var (
@ -858,11 +881,8 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) {
return
}
if p.Opcode != proto.OpBackupWrite {
if s.isWriteOpOfProtoVersionForbidden(p.ProtoVersion) {
err = fmt.Errorf("%v %v", storage.WriteOpOfProtoVerForbidden, p.ProtoVersion)
return
}
if err = s.checkForbidWriteOpOfProtoVer0(p, partition); err != nil {
return
}
shallDegrade := p.ShallDegrade()
@ -1026,8 +1046,7 @@ func (s *DataNode) handleRandomWritePacket(p *repl.Packet) {
return
}
if s.isWriteOpOfProtoVersionForbidden(p.ProtoVersion) {
err = fmt.Errorf("%v %v", storage.WriteOpOfProtoVerForbidden, p.ProtoVersion)
if err = s.checkForbidWriteOpOfProtoVer0(p, partition); err != nil {
return
}

3
datanode/wrap_operator_test.go
View File

@ -58,7 +58,8 @@ func newDpForOperatorTest(t *testing.T, dn *DataNode) (dp *DataPartition) {
disk: newDiskForOperatorTest(t, dn),
extentStore: newExtentStoreForOperatorTest(t),
config: &dataPartitionCfg{
Forbidden: false,
Forbidden: false,
ForbidWriteOpOfProtoVer0: false,
},
partitionSize: 1 * util.TB,
dataNode: dn,

13
fsck/cmd/gc.go
View File

@ -663,13 +663,12 @@ func newGetDpExtentsCmd() *cobra.Command {
log.LogInfof("beforeTime: %v", beforeTime)
currentTimeUnix := time.Now().Unix()
beforeTimeUnix := t.Unix()
//TODO:tangjingyu comment this for test. must uncomment before release
//threeHours := int64(3600 * 3)
//
//if currentTimeUnix-beforeTimeUnix < threeHours {
// slog.Fatalf("beforeTime should be at least 3 hours earlier than current time %v %v", currentTimeUnix, beforeTimeUnix)
// return
//}
threeHours := int64(3600 * 3)
if currentTimeUnix-beforeTimeUnix < threeHours {
slog.Fatalf("beforeTime should be at least 3 hours earlier than current time %v %v", currentTimeUnix, beforeTimeUnix)
return
}
oneMinute := int64(1)
if currentTimeUnix-beforeTimeUnix < oneMinute {
slog.Fatalf("beforeTime should be at least 1 minute earlier than current time %v %v", currentTimeUnix, beforeTimeUnix)

58
master/api_args_parse.go
View File

@ -394,32 +394,33 @@ func extractBoolWithDefault(r *http.Request, key string, def bool) (val bool, er
}
type updateVolReq struct {
name string
authKey string
capacity uint64
deleteLockTime int64
followerRead bool
authenticate bool
enablePosixAcl bool
enableTransaction proto.TxOpMask
txTimeout int64
txConflictRetryNum int64
txConflictRetryInterval int64
txOpLimit int
zoneName string
description string
dpSelectorName string
dpSelectorParm string
replicaNum int
coldArgs *coldVolArgs
dpReadOnlyWhenVolFull bool
enableQuota bool
crossZone bool
trashInterval int64
enableAutoDpMetaRepair bool
accessTimeValidInterval int64
enablePersistAccessTime bool
volStorageClass uint32
name string
authKey string
capacity uint64
deleteLockTime int64
followerRead bool
authenticate bool
enablePosixAcl bool
enableTransaction proto.TxOpMask
txTimeout int64
txConflictRetryNum int64
txConflictRetryInterval int64
txOpLimit int
zoneName string
description string
dpSelectorName string
dpSelectorParm string
replicaNum int
coldArgs *coldVolArgs
dpReadOnlyWhenVolFull bool
enableQuota bool
crossZone bool
trashInterval int64
enableAutoDpMetaRepair bool
accessTimeValidInterval int64
enablePersistAccessTime bool
volStorageClass uint32
forbidWriteOpOfProtoVer0 bool
}
func parseColdVolUpdateArgs(r *http.Request, vol *Vol) (args *coldVolArgs, err error) {
@ -582,6 +583,11 @@ func parseVolUpdateReq(r *http.Request, vol *Vol, req *updateVolReq) (err error)
return
}
if req.forbidWriteOpOfProtoVer0, err = extractBoolWithDefault(r, forbidWriteOpOfProtoVersion0, vol.ForbidWriteOpOfProtoVer0.Load()); err != nil {
return
}
log.LogDebugf("[parseVolUpdateReq] vol(%v) forbidWriteOpOfProtoVer0: %v", vol.Name, req.forbidWriteOpOfProtoVer0)
req.dpSelectorName = r.FormValue(dpSelectorNameKey)
req.dpSelectorParm = r.FormValue(dpSelectorParmKey)

78
master/api_service.go
View File

@ -1003,6 +1003,15 @@ func (m *Server) getIPAddr(w http.ResponseWriter, r *http.Request) {
dirChildrenNumLimit := atomic.LoadUint32(&m.cluster.cfg.DirChildrenNumLimit)
dpMaxRepairErrCnt := atomic.LoadUint64(&m.cluster.cfg.DpMaxRepairErrCnt)
volsForbidWriteOpOfProtoVer0 := make([]string, 0)
m.cluster.volMutex.RLock()
for _, vol := range m.cluster.vols {
if vol.ForbidWriteOpOfProtoVer0.Load() {
volsForbidWriteOpOfProtoVer0 = append(volsForbidWriteOpOfProtoVer0, vol.Name)
}
}
m.cluster.volMutex.RUnlock()
cInfo := &proto.ClusterInfo{
Cluster: m.cluster.Name,
MetaNodeDeleteBatchCount: batchCount,
@ -1012,14 +1021,17 @@ func (m *Server) getIPAddr(w http.ResponseWriter, r *http.Request) {
DpMaxRepairErrCnt: dpMaxRepairErrCnt,
DirChildrenNumLimit: dirChildrenNumLimit,
// Ip: strings.Split(r.RemoteAddr, ":")[0],
Ip: iputil.RealIP(r),
EbsAddr: m.bStoreAddr,
ServicePath: m.servicePath,
ClusterUuid: m.cluster.clusterUuid,
ClusterUuidEnable: m.cluster.clusterUuidEnable,
ClusterEnableSnapshot: m.cluster.cfg.EnableSnapshot,
ForbidWriteOpOfProtoVer0: m.cluster.cfg.forbidWriteOpOfProtoVer0,
Ip: iputil.RealIP(r),
EbsAddr: m.bStoreAddr,
ServicePath: m.servicePath,
ClusterUuid: m.cluster.clusterUuid,
ClusterUuidEnable: m.cluster.clusterUuidEnable,
ClusterEnableSnapshot: m.cluster.cfg.EnableSnapshot,
ForbidWriteOpOfProtoVer0: m.cluster.cfg.forbidWriteOpOfProtoVer0,
VolsForbidWriteOpOfProtoVer0: volsForbidWriteOpOfProtoVer0,
}
log.LogDebugf("[getIPAddr] %v VolsForbidWriteOpOfProtoVer0: %v",
len(cInfo.VolsForbidWriteOpOfProtoVer0), cInfo.VolsForbidWriteOpOfProtoVer0)
sendOkReply(w, r, newSuccessHTTPReply(cInfo))
}
@ -2425,6 +2437,7 @@ func (m *Server) updateVol(w http.ResponseWriter, r *http.Request) {
newArgs.dpReadOnlyWhenVolFull = req.dpReadOnlyWhenVolFull
newArgs.enableAutoDpMetaRepair = req.enableAutoDpMetaRepair
newArgs.volStorageClass = req.volStorageClass
newArgs.forbidWriteOpOfProtoVer0 = req.forbidWriteOpOfProtoVer0
log.LogWarnf("[updateVolOut] name [%s], z1 [%s], z2[%s] replicaNum[%v]", req.name, req.zoneName, vol.Name, req.replicaNum)
if err = m.cluster.updateVol(req.name, req.authKey, newArgs); err != nil {
@ -2976,9 +2989,10 @@ func newSimpleView(vol *Vol) (view *proto.SimpleVolView) {
AccessTimeInterval: vol.AccessTimeValidInterval,
EnablePersistAccessTime: vol.EnablePersistAccessTime,
AllowedStorageClass: vol.allowedStorageClass,
VolStorageClass: vol.volStorageClass,
CacheDpStorageClass: vol.cacheDpStorageClass,
AllowedStorageClass: vol.allowedStorageClass,
VolStorageClass: vol.volStorageClass,
CacheDpStorageClass: vol.cacheDpStorageClass,
ForbidWriteOpOfProtoVer0: vol.ForbidWriteOpOfProtoVer0.Load(),
}
view.AllowedStorageClass = make([]uint32, len(vol.allowedStorageClass))
copy(view.AllowedStorageClass, vol.allowedStorageClass)
@ -5572,34 +5586,38 @@ func (m *Server) getMetaPartition(w http.ResponseWriter, r *http.Request) {
MaxInode: mp.Replicas[i].MaxInodeID,
}
}
forbidden := true
forbidWriteOpOfProtoVer0 := false
vol, err := m.cluster.getVol(mp.volName)
if err == nil {
forbidden = vol.Forbidden
forbidWriteOpOfProtoVer0 = vol.ForbidWriteOpOfProtoVer0.Load()
} else {
log.LogErrorf("action[getMetaPartition]failed to get volume %v, err %v", mp.volName, err)
}
mpInfo := &proto.MetaPartitionInfo{
PartitionID: mp.PartitionID,
Start: mp.Start,
End: mp.End,
VolName: mp.volName,
MaxInodeID: mp.MaxInodeID,
InodeCount: mp.InodeCount,
DentryCount: mp.DentryCount,
Replicas: replicas,
ReplicaNum: mp.ReplicaNum,
Status: mp.Status,
IsRecover: mp.IsRecover,
Hosts: mp.Hosts,
Peers: mp.Peers,
Zones: zones,
NodeSets: nodeSets,
MissNodes: mp.MissNodes,
OfflinePeerID: mp.OfflinePeerID,
LoadResponse: mp.LoadResponse,
Forbidden: forbidden,
StatByStorageClass: mp.StatByStorageClass,
PartitionID: mp.PartitionID,
Start: mp.Start,
End: mp.End,
VolName: mp.volName,
MaxInodeID: mp.MaxInodeID,
InodeCount: mp.InodeCount,
DentryCount: mp.DentryCount,
Replicas: replicas,
ReplicaNum: mp.ReplicaNum,
Status: mp.Status,
IsRecover: mp.IsRecover,
Hosts: mp.Hosts,
Peers: mp.Peers,
Zones: zones,
NodeSets: nodeSets,
MissNodes: mp.MissNodes,
OfflinePeerID: mp.OfflinePeerID,
LoadResponse: mp.LoadResponse,
Forbidden: forbidden,
StatByStorageClass: mp.StatByStorageClass,
ForbidWriteOpOfProtoVer0: forbidWriteOpOfProtoVer0,
}
return mpInfo
}

14
master/cluster.go
View File

@ -847,6 +847,10 @@ func (c *Cluster) checkDataNodeHeartbeat() {
if vol.dpRepairBlockSize != proto.DefaultDpRepairBlockSize {
hbReq.VolDpRepairBlockSize[vol.Name] = vol.dpRepairBlockSize
}
if vol.ForbidWriteOpOfProtoVer0.Load() {
hbReq.VolsForbidWriteOpOfProtoVer0 = append(hbReq.VolsForbidWriteOpOfProtoVer0, vol.Name)
}
}
tasks = append(tasks, task)
return true
@ -878,6 +882,9 @@ func (c *Cluster) checkMetaNodeHeartbeat() {
if vol.Forbidden {
hbReq.ForbiddenVols = append(hbReq.ForbiddenVols, vol.Name)
}
if vol.ForbidWriteOpOfProtoVer0.Load() {
hbReq.VolsForbidWriteOpOfProtoVer0 = append(hbReq.VolsForbidWriteOpOfProtoVer0, vol.Name)
}
spaceInfo := vol.uidSpaceManager.getSpaceOp()
hbReq.UidLimitInfo = append(hbReq.UidLimitInfo, spaceInfo...)
@ -1883,6 +1890,7 @@ func (c *Cluster) createDataPartition(volName string, preload *DataPartitionPreL
if diskPath, err = c.syncCreateDataPartitionToDataNode(host, vol.dataPartitionSize,
dp, dp.Peers, dp.Hosts, proto.NormalCreateDataPartition, dp.PartitionType, false, false); err != nil {
log.LogErrorf("[createDataPartition] %v", err)
errChannel <- err
return
}
@ -1954,10 +1962,11 @@ func (c *Cluster) syncCreateDataPartitionToDataNode(host string, size uint64, dp
task = dp.createTaskToCreateDataPartition(host, size, peers, hosts, createType, partitionType, dataNode.getDecommissionedDisks())
}
if task == nil {
err = errors.NewErrorf("action[syncCreateDataPartitionToDataNode] dp[%v] create task for creating data partition failed",
dp.decommissionInfo())
err = errors.NewErrorf("action[syncCreateDataPartitionToDataNode] dp[%v] meditType(%v) create task for creating data partition failed",
dp.decommissionInfo(), proto.MediaTypeString(dp.MediaType))
return
}
var resp *proto.Packet
if resp, err = dataNode.TaskManager.syncSendAdminTask(task); err != nil {
// data node is not alive or other process error
@ -3130,6 +3139,7 @@ func (c *Cluster) createDataReplica(dp *DataPartition, addPeer proto.Peer, ignor
diskPath, err := c.syncCreateDataPartitionToDataNode(addPeer.Addr, vol.dataPartitionSize,
dp, peers, hosts, proto.DecommissionedCreateDataPartition, dp.PartitionType, true, ignoreDecommissionDisk)
if err != nil {
log.LogErrorf("[createDataReplica] %v", err)
return
}

83
master/const.go
View File

@ -190,47 +190,48 @@ const (
)
const (
defaultFaultDomainZoneCnt = 3
defaultNormalCrossZoneCnt = 3
defaultInitMetaPartitionCount = 3
defaultMaxInitMetaPartitionCount = 100
defaultMaxMetaPartitionInodeID uint64 = 1<<63 - 1
defaultMetaPartitionInodeIDStep uint64 = 1 << 22
defaultMetaNodeReservedMem uint64 = 1 << 30
runtimeStackBufSize = 4096
spaceAvailableRate = 0.90
defaultNodeSetCapacity = 18
minNumOfRWDataPartitions = 10
intervalToCheckMissingReplica = 600
intervalToWarnDataPartition = 600
intervalToLoadDataPartition = 12 * 60 * 60
defaultInitDataPartitionCnt = 10
maxInitDataPartitionCnt = 200
volExpansionRatio = 0.1
maxNumberOfDataPartitionsForExpansion = 100
EmptyCrcValue uint32 = 4045511210
DefaultZoneName = proto.DefaultZoneName
retrySendSyncTaskInternal = 3 * time.Second
defaultRangeOfCountDifferencesAllowed = 50
defaultMinusOfMaxInodeID = 1000
defaultNodeSetGrpBatchCnt = 3
defaultMigrateDpCnt = 50
defaultMigrateMpCnt = 3
defaultMaxReplicaCnt = 16
defaultIopsRLimit uint64 = 1 << 35
defaultIopsWLimit uint64 = 1 << 35
defaultFlowWLimit uint64 = 1 << 35
defaultFlowRLimit uint64 = 1 << 35
defaultLimitTypeCnt = 4
defaultClientTriggerHitCnt = 1
defaultClientReqPeriodSeconds = 1
defaultMaxQuotaNumPerVol = 100
defaultVolDelayDeleteTimeHour = 48
defaultMarkDiskBrokenThreshold = 0 // decommission all dp from disk
defaultEnableDpMetaRepair = false
defaultAutoDpMetaRepairPallarelCnt = 100
defaultAutoDecommissionDiskInterval = 10 * time.Second
maxMpCreationCount = 10
defaultFaultDomainZoneCnt = 3
defaultNormalCrossZoneCnt = 3
defaultInitMetaPartitionCount = 3
defaultMaxInitMetaPartitionCount = 100
defaultMaxMetaPartitionInodeID uint64 = 1<<63 - 1
defaultMetaPartitionInodeIDStep uint64 = 1 << 22
defaultMetaNodeReservedMem uint64 = 1 << 30
runtimeStackBufSize = 4096
spaceAvailableRate = 0.90
defaultNodeSetCapacity = 18
minNumOfRWDataPartitions = 10
intervalToCheckMissingReplica = 600
intervalToWarnDataPartition = 600
intervalToLoadDataPartition = 12 * 60 * 60
defaultInitDataPartitionCnt = 10
maxInitDataPartitionCnt = 200
volExpansionRatio = 0.1
maxNumberOfDataPartitionsForExpansion = 100
EmptyCrcValue uint32 = 4045511210
DefaultZoneName = proto.DefaultZoneName
retrySendSyncTaskInternal = 3 * time.Second
defaultRangeOfCountDifferencesAllowed = 50
defaultMinusOfMaxInodeID = 1000
defaultNodeSetGrpBatchCnt = 3
defaultMigrateDpCnt = 50
defaultMigrateMpCnt = 3
defaultMaxReplicaCnt = 16
defaultIopsRLimit uint64 = 1 << 35
defaultIopsWLimit uint64 = 1 << 35
defaultFlowWLimit uint64 = 1 << 35
defaultFlowRLimit uint64 = 1 << 35
defaultLimitTypeCnt = 4
defaultClientTriggerHitCnt = 1
defaultClientReqPeriodSeconds = 1
defaultMaxQuotaNumPerVol = 100
defaultVolDelayDeleteTimeHour = 48
defaultMarkDiskBrokenThreshold = 0 // decommission all dp from disk
defaultEnableDpMetaRepair = false
defaultAutoDpMetaRepairPallarelCnt = 100
defaultAutoDecommissionDiskInterval = 10 * time.Second
maxMpCreationCount = 10
defaultVolForbidWriteOpOfProtoVersion0 = false
)
const (

3
master/data_partition.go
View File

@ -979,9 +979,11 @@ func (partition *DataPartition) buildDpInfo(c *Cluster) *proto.DataPartitionInfo
}
forbidden := true
forbidWriteOpOfProtoVer0 := false
vol, err := c.getVol(partition.VolName)
if err == nil {
forbidden = vol.Forbidden
forbidWriteOpOfProtoVer0 = vol.ForbidWriteOpOfProtoVer0.Load()
} else {
log.LogErrorf("action[buildDpInfo]failed to get volume %v, err %v", partition.VolName, err)
}
@ -1009,6 +1011,7 @@ func (partition *DataPartition) buildDpInfo(c *Cluster) *proto.DataPartitionInfo
SingleDecommissionStatus: partition.GetSpecialReplicaDecommissionStep(),
Forbidden: forbidden,
MediaType: partition.MediaType,
ForbidWriteOpOfProtoVer0: forbidWriteOpOfProtoVer0,
}
}

12
master/metadata_fsm_op.go
View File

@ -336,9 +336,10 @@ type volValue struct {
DpRepairBlockSize uint64
EnableAutoMetaRepair bool
VolStorageClass uint32
AllowedStorageClass []uint32
CacheDpStorageClass uint32
VolStorageClass uint32
AllowedStorageClass []uint32
CacheDpStorageClass uint32
ForbidWriteOpOfProtoVer0 bool
}
func (v *volValue) Bytes() (raw []byte, err error) {
@ -411,8 +412,9 @@ func newVolValue(vol *Vol) (vv *volValue) {
AccessTimeInterval: vol.AccessTimeValidInterval,
EnablePersistAccessTime: vol.EnablePersistAccessTime,
VolStorageClass: vol.volStorageClass,
CacheDpStorageClass: vol.cacheDpStorageClass,
VolStorageClass: vol.volStorageClass,
CacheDpStorageClass: vol.cacheDpStorageClass,
ForbidWriteOpOfProtoVer0: vol.ForbidWriteOpOfProtoVer0.Load(),
}
vv.AllowedStorageClass = make([]uint32, len(vol.allowedStorageClass))
copy(vv.AllowedStorageClass, vol.allowedStorageClass)

129
master/vol.go
View File

@ -33,32 +33,34 @@ import (
)
type VolVarargs struct {
zoneName string
description string
capacity uint64 // GB
deleteLockTime int64 // h
followerRead bool
authenticate bool
dpSelectorName string
dpSelectorParm string
coldArgs *coldVolArgs
dpReplicaNum uint8
enablePosixAcl bool
dpReadOnlyWhenVolFull bool
enableQuota bool
enableTransaction proto.TxOpMask
txTimeout int64
txConflictRetryNum int64
txConflictRetryInterval int64
txOpLimit int
trashInterval int64
crossZone bool
accessTimeInterval int64
enableAutoDpMetaRepair bool
accessTimeValidInterval int64
enablePersistAccessTime bool
volStorageClass uint32
allowedStorageClass []uint32
zoneName string
description string
capacity uint64 // GB
deleteLockTime int64 // h
followerRead bool
authenticate bool
dpSelectorName string
dpSelectorParm string
coldArgs *coldVolArgs
domainId uint64
dpReplicaNum uint8
enablePosixAcl bool
dpReadOnlyWhenVolFull bool
enableQuota bool
enableTransaction proto.TxOpMask
txTimeout int64
txConflictRetryNum int64
txConflictRetryInterval int64
txOpLimit int
trashInterval int64
crossZone bool
accessTimeInterval int64
enableAutoDpMetaRepair bool
accessTimeValidInterval int64
enablePersistAccessTime bool
volStorageClass uint32
allowedStorageClass []uint32
forbidWriteOpOfProtoVer0 bool
}
type CacheSubItem struct {
@ -133,16 +135,17 @@ type Vol struct {
mpsCache []byte
viewCache []byte
NeedToLowerReplica bool
FollowerRead bool
enableQuota bool
DisableAuditLog bool
DpReadOnlyWhenVolFull bool // only if this switch is on, all dp becomes readonly when vol is full
ReadOnlyForVolFull bool // only if the switch DpReadOnlyWhenVolFull is on, mark vol is readonly when is full
AccessTimeInterval int64
EnablePersistAccessTime bool
AccessTimeValidInterval int64
EnableAutoMetaRepair atomicutil.Bool
NeedToLowerReplica bool
FollowerRead bool
enableQuota bool
DisableAuditLog bool
DpReadOnlyWhenVolFull bool // only if this switch is on, all dp becomes readonly when vol is full
ReadOnlyForVolFull bool // only if the switch DpReadOnlyWhenVolFull is on, mark vol is readonly when is full
AccessTimeInterval int64
EnablePersistAccessTime bool
AccessTimeValidInterval int64
EnableAutoMetaRepair atomicutil.Bool
ForbidWriteOpOfProtoVer0 atomicutil.Bool
TopoSubItem
CacheSubItem
@ -240,6 +243,7 @@ func newVol(vv volValue) (vol *Vol) {
vol.volStorageClass = vv.VolStorageClass
vol.cacheDpStorageClass = vv.CacheDpStorageClass
vol.StatByStorageClass = make([]*proto.StatOfStorageClass, 0)
vol.ForbidWriteOpOfProtoVer0.Store(defaultVolForbidWriteOpOfProtoVersion0)
return
}
@ -276,6 +280,7 @@ func newVolFromVolValue(vv *volValue) (vol *Vol) {
if vol.AccessTimeValidInterval == 0 {
vol.AccessTimeValidInterval = proto.DefaultAccessTimeValidInterval
}
vol.ForbidWriteOpOfProtoVer0.Store(vv.ForbidWriteOpOfProtoVer0)
return vol
}
@ -1726,6 +1731,7 @@ func setVolFromArgs(args *VolVarargs, vol *Vol) {
vol.EnablePersistAccessTime = args.enablePersistAccessTime
vol.volStorageClass = args.volStorageClass
vol.allowedStorageClass = append([]uint32{}, args.allowedStorageClass...)
vol.ForbidWriteOpOfProtoVer0.Store(args.forbidWriteOpOfProtoVer0)
}
func getVolVarargs(vol *Vol) *VolVarargs {
@ -1745,31 +1751,32 @@ func getVolVarargs(vol *Vol) *VolVarargs {
}
return &VolVarargs{
zoneName: vol.zoneName,
crossZone: vol.crossZone,
description: vol.description,
capacity: vol.Capacity,
deleteLockTime: vol.DeleteLockTime,
followerRead: vol.FollowerRead,
authenticate: vol.authenticate,
dpSelectorName: vol.dpSelectorName,
dpSelectorParm: vol.dpSelectorParm,
enablePosixAcl: vol.enablePosixAcl,
enableQuota: vol.enableQuota,
dpReplicaNum: vol.dpReplicaNum,
enableTransaction: vol.enableTransaction,
txTimeout: vol.txTimeout,
txConflictRetryNum: vol.txConflictRetryNum,
txConflictRetryInterval: vol.txConflictRetryInterval,
txOpLimit: vol.txOpLimit,
coldArgs: args,
dpReadOnlyWhenVolFull: vol.DpReadOnlyWhenVolFull,
accessTimeValidInterval: vol.AccessTimeValidInterval,
trashInterval: vol.TrashInterval,
enablePersistAccessTime: vol.EnablePersistAccessTime,
enableAutoDpMetaRepair: vol.EnableAutoMetaRepair.Load(),
volStorageClass: vol.volStorageClass,
allowedStorageClass: append([]uint32{}, vol.allowedStorageClass...),
zoneName: vol.zoneName,
crossZone: vol.crossZone,
description: vol.description,
capacity: vol.Capacity,
deleteLockTime: vol.DeleteLockTime,
followerRead: vol.FollowerRead,
authenticate: vol.authenticate,
dpSelectorName: vol.dpSelectorName,
dpSelectorParm: vol.dpSelectorParm,
enablePosixAcl: vol.enablePosixAcl,
enableQuota: vol.enableQuota,
dpReplicaNum: vol.dpReplicaNum,
enableTransaction: vol.enableTransaction,
txTimeout: vol.txTimeout,
txConflictRetryNum: vol.txConflictRetryNum,
txConflictRetryInterval: vol.txConflictRetryInterval,
txOpLimit: vol.txOpLimit,
coldArgs: args,
dpReadOnlyWhenVolFull: vol.DpReadOnlyWhenVolFull,
accessTimeValidInterval: vol.AccessTimeValidInterval,
trashInterval: vol.TrashInterval,
enablePersistAccessTime: vol.EnablePersistAccessTime,
enableAutoDpMetaRepair: vol.EnableAutoMetaRepair.Load(),
volStorageClass: vol.volStorageClass,
allowedStorageClass: append([]uint32{}, vol.allowedStorageClass...),
forbidWriteOpOfProtoVer0: vol.ForbidWriteOpOfProtoVer0.Load(),
}
}

2
metanode/inode.go
View File

@ -2516,7 +2516,7 @@ func (i *Inode) updateStorageClass(storageClass uint32, isCache, isMigration boo
if i.StorageClass == proto.StorageClass_Unspecified {
i.StorageClass = storageClass
} else if i.StorageClass != storageClass {
return errors.New(fmt.Sprintf("storageClass %v not equal to inode.storageClass %v",
return errors.New(fmt.Sprintf("req.storageClass(%v) not equal to inode.storageClass(%v)",
storageClass, i.StorageClass))
}
}

17
metanode/manager.go
View File

@ -28,6 +28,7 @@ import (
"time"
"github.com/cubefs/cubefs/cmd/common"
"github.com/cubefs/cubefs/datanode/storage"
"github.com/cubefs/cubefs/proto"
"github.com/cubefs/cubefs/raftstore"
"github.com/cubefs/cubefs/util"
@ -189,12 +190,22 @@ func (m *metadataManager) getPacketLabels(p *Packet) (labels map[string]string)
return
}
func (m *metadataManager) isWriteOpOfProtoVersionForbidden(pktProtoVersion uint32) (forbidden bool) {
func (m *metadataManager) checkForbidWriteOpOfProtoVer0(pktProtoVersion uint32, mpForbidWriteOpOfProtoVer0 bool) (err error) {
if pktProtoVersion != proto.PacketProtoVersion0 {
return false
return nil
}
return m.metaNode.forbidWriteOpOfProtoVer0
if m.metaNode.nodeForbidWriteOpOfProtoVer0 {
err = fmt.Errorf("%v %v", storage.ClusterForbidWriteOpOfProtoVer, pktProtoVersion)
return
}
if mpForbidWriteOpOfProtoVer0 {
err = fmt.Errorf("%v %v", storage.VolForbidWriteOpOfProtoVer, pktProtoVersion)
return
}
return nil
}
// HandleMetadataOperation handles the metadata operations.

53
metanode/manager_op.go
View File

@ -49,6 +49,34 @@ func (m *metadataManager) checkFollowerRead(volNames []string, partition MetaPar
partition.SetFollowerRead(false)
}
func (m *metadataManager) checkVolForbidWriteOpOfProtoVer0(partition MetaPartition) {
mpVolName := partition.GetVolName()
oldVal := partition.IsForbidWriteOpOfProtoVer0()
VolsForbidWriteOpOfProtoVer0 := m.metaNode.VolsForbidWriteOpOfProtoVer0
if _, ok := VolsForbidWriteOpOfProtoVer0[mpVolName]; ok {
partition.SetForbidWriteOpOfProtoVer0(true)
} else {
partition.SetForbidWriteOpOfProtoVer0(false)
}
newVal := partition.IsForbidWriteOpOfProtoVer0()
if oldVal != newVal {
log.LogInfof("[checkVolForbidWriteOpOfProtoVer0] vol(%v) mpId(%v) IsForbidWriteOpOfProtoVer0 change to %v",
mpVolName, partition.GetBaseConfig().PartitionId, newVal)
}
return
}
func (m *metadataManager) isVolForbidWriteOpOfProtoVer0(volName string) (forbid bool) {
VolsForbidWriteOpOfProtoVer0 := m.metaNode.VolsForbidWriteOpOfProtoVer0
if _, ok := VolsForbidWriteOpOfProtoVer0[volName]; ok {
forbid = true
} else {
forbid = false
}
return
}
func (m *metadataManager) checkForbiddenVolume(volNames []string, partition MetaPartition) {
volName := partition.GetVolName()
for _, name := range volNames {
@ -84,6 +112,7 @@ func (m *metadataManager) opMasterHeartbeat(conn net.Conn, p *Packet,
adminTask = &proto.AdminTask{
Request: req,
}
volsForbidWriteOpOfProtoVer0 = make(map[string]struct{})
)
start := time.Now()
go func() {
@ -95,12 +124,21 @@ func (m *metadataManager) opMasterHeartbeat(conn net.Conn, p *Packet,
goto end
}
m.fileStatsEnable = req.FileStatsEnable
if m.metaNode.forbidWriteOpOfProtoVer0 != req.NotifyForbidWriteOpOfProtoVer0 {
log.LogWarnf("[opMasterHeartbeat] change forbidWriteOpOfProtoVer0, old(%v) new(%v)",
m.metaNode.forbidWriteOpOfProtoVer0, req.NotifyForbidWriteOpOfProtoVer0)
m.metaNode.forbidWriteOpOfProtoVer0 = req.NotifyForbidWriteOpOfProtoVer0
if m.metaNode.nodeForbidWriteOpOfProtoVer0 != req.NotifyForbidWriteOpOfProtoVer0 {
log.LogWarnf("[opMasterHeartbeat] change nodeForbidWriteOpOfProtoVer0, old(%v) new(%v)",
m.metaNode.nodeForbidWriteOpOfProtoVer0, req.NotifyForbidWriteOpOfProtoVer0)
m.metaNode.nodeForbidWriteOpOfProtoVer0 = req.NotifyForbidWriteOpOfProtoVer0
}
for _, vol := range req.VolsForbidWriteOpOfProtoVer0 {
if _, ok := volsForbidWriteOpOfProtoVer0[vol]; !ok {
volsForbidWriteOpOfProtoVer0[vol] = struct{}{}
}
}
m.metaNode.VolsForbidWriteOpOfProtoVer0 = volsForbidWriteOpOfProtoVer0
log.LogDebugf("[opMasterHeartbeat] from master, volumes forbid write operate of proto version-0: %v",
req.VolsForbidWriteOpOfProtoVer0)
// collect memory info
resp.Total = configTotalMem
resp.Used, err = util.GetProcessMemory(os.Getpid())
@ -114,6 +152,7 @@ func (m *metadataManager) opMasterHeartbeat(conn net.Conn, p *Packet,
m.Range(true, func(id uint64, partition MetaPartition) bool {
m.checkFollowerRead(req.FLReadVols, partition)
m.checkForbiddenVolume(req.ForbiddenVols, partition)
m.checkVolForbidWriteOpOfProtoVer0(partition)
m.checkDisableAuditLogVolume(req.DisableAuditVols, partition)
partition.SetUidLimit(req.UidLimitInfo)
partition.SetTxInfo(req.TxInfo)
@ -154,7 +193,7 @@ func (m *metadataManager) opMasterHeartbeat(conn net.Conn, p *Packet,
return true
})
resp.ZoneName = m.zoneName
resp.ReceivedForbidWriteOpOfProtoVer0 = m.metaNode.forbidWriteOpOfProtoVer0
resp.ReceivedForbidWriteOpOfProtoVer0 = m.metaNode.nodeForbidWriteOpOfProtoVer0
resp.Status = proto.TaskSucceeds
end:
adminTask.Request = nil
@ -1374,8 +1413,8 @@ func (m *metadataManager) opMetaExtentsTruncate(conn net.Conn, p *Packet,
return
}
if m.isWriteOpOfProtoVersionForbidden(p.ProtoVersion) {
err = fmt.Errorf("%v %v", storage.WriteOpOfProtoVerForbidden, p.ProtoVersion)
if err = m.checkForbidWriteOpOfProtoVer0(p.ProtoVersion, mp.IsForbidWriteOpOfProtoVer0()); err != nil {
log.LogWarnf("[opMetaExtentsTruncate] reqId(%v) mpId(%v) ino(%v) err: %v", p.ReqID, req.PartitionID, req.Inode, err)
p.PacketErrorWithBody(proto.OpWriteOpOfProtoVerForbidden, ([]byte)(err.Error()))
m.respondToClientWithVer(conn, p)
return

73
metanode/metanode.go
View File

@ -52,31 +52,32 @@ var (
// The MetaNode manages the dentry and inode information of the meta partitions on a meta node.
// The data consistency is ensured by Raft.
type MetaNode struct {
nodeId uint64
listen string
bindIp bool
metadataDir string // root dir of the metaNode
raftDir string // root dir of the raftStore log
metadataManager MetadataManager
localAddr string
clusterId string
raftStore raftstore.RaftStore
raftHeartbeatPort string
raftReplicatePort string
raftRetainLogs uint64
raftSyncSnapFormatVersion uint32 // format version of snapshot that raft leader sent to follower
zoneName string
httpStopC chan uint8
smuxStopC chan uint8
metrics *MetaNodeMetrics
tickInterval int
raftRecvBufSize int
connectionCnt int64
clusterUuid string
clusterUuidEnable bool
clusterEnableSnapshot bool
serviceIDKey string
forbidWriteOpOfProtoVer0 bool
nodeId uint64
listen string
bindIp bool
metadataDir string // root dir of the metaNode
raftDir string // root dir of the raftStore log
metadataManager MetadataManager
localAddr string
clusterId string
raftStore raftstore.RaftStore
raftHeartbeatPort string
raftReplicatePort string
raftRetainLogs uint64
raftSyncSnapFormatVersion uint32 // format version of snapshot that raft leader sent to follower
zoneName string
httpStopC chan uint8
smuxStopC chan uint8
metrics *MetaNodeMetrics
tickInterval int
raftRecvBufSize int
connectionCnt int64
clusterUuid string
clusterUuidEnable bool
clusterEnableSnapshot bool
serviceIDKey string
nodeForbidWriteOpOfProtoVer0 bool // whether forbid by node granularity,
VolsForbidWriteOpOfProtoVer0 map[string]struct{} // whether forbid by volume granularity,
control common.Control
}
@ -488,11 +489,23 @@ func (m *MetaNode) register() (err error) {
clusterEnableSnapshot = m.clusterEnableSnapshot
m.clusterId = clusterInfo.Cluster
nodeAddress = m.localAddr + ":" + m.listen
m.forbidWriteOpOfProtoVer0 = clusterInfo.ForbidWriteOpOfProtoVer0
forbiddenWriteOpVerMsg := fmt.Sprintf("[register] got from master forbidWriteOpOfProtoVer0: %v",
m.forbidWriteOpOfProtoVer0)
log.LogInfo(forbiddenWriteOpVerMsg)
syslog.Printf("%v \n", forbiddenWriteOpVerMsg)
m.nodeForbidWriteOpOfProtoVer0 = clusterInfo.ForbidWriteOpOfProtoVer0
nodeForbidWriteOpOfProtoVerMsg := fmt.Sprintf("[register] got from master, node forbid write Operate Of proto version-0: %v",
m.nodeForbidWriteOpOfProtoVer0)
log.LogInfo(nodeForbidWriteOpOfProtoVerMsg)
syslog.Printf("%v \n", nodeForbidWriteOpOfProtoVerMsg)
volsForbidWriteOpOfProtoVer0 := make(map[string]struct{})
for _, vol := range clusterInfo.VolsForbidWriteOpOfProtoVer0 {
if _, ok := volsForbidWriteOpOfProtoVer0[vol]; !ok {
volsForbidWriteOpOfProtoVer0[vol] = struct{}{}
}
}
m.VolsForbidWriteOpOfProtoVer0 = volsForbidWriteOpOfProtoVer0
volsForbidWriteOpVerMsg := fmt.Sprintf("action[registerToMaster] from master, volumes forbid write operate of proto version-0: %v",
clusterInfo.VolsForbidWriteOpOfProtoVer0)
log.LogInfo(volsForbidWriteOpVerMsg)
syslog.Printf("%v\n", volsForbidWriteOpVerMsg)
step++
}

50
metanode/partition.go
View File

@ -82,24 +82,25 @@ type MetaMultiSnapshotInfo struct {
// MetaPartitionConfig is used to create a meta partition.
type MetaPartitionConfig struct {
// Identity for raftStore group. RaftStore nodes in the same raftStore group must have the same groupID.
PartitionId uint64 `json:"partition_id"`
VolName string `json:"vol_name"`
Start uint64 `json:"start"` // Minimal Inode ID of this range. (Required during initialization)
End uint64 `json:"end"` // Maximal Inode ID of this range. (Required during initialization)
PartitionType int `json:"partition_type"`
Peers []proto.Peer `json:"peers"` // Peers information of the raftStore
Cursor uint64 `json:"-"` // Cursor ID of the inode that have been assigned
UniqId uint64 `json:"-"`
NodeId uint64 `json:"-"`
RootDir string `json:"-"`
VerSeq uint64 `json:"ver_seq"`
BeforeStart func() `json:"-"`
AfterStart func() `json:"-"`
BeforeStop func() `json:"-"`
AfterStop func() `json:"-"`
RaftStore raftstore.RaftStore `json:"-"`
ConnPool *util.ConnectPool `json:"-"`
Forbidden bool `json:"-"`
PartitionId uint64 `json:"partition_id"`
VolName string `json:"vol_name"`
Start uint64 `json:"start"` // Minimal Inode ID of this range. (Required during initialization)
End uint64 `json:"end"` // Maximal Inode ID of this range. (Required during initialization)
PartitionType int `json:"partition_type"`
Peers []proto.Peer `json:"peers"` // Peers information of the raftStore
Cursor uint64 `json:"-"` // Cursor ID of the inode that have been assigned
UniqId uint64 `json:"-"`
NodeId uint64 `json:"-"`
RootDir string `json:"-"`
VerSeq uint64 `json:"ver_seq"`
BeforeStart func() `json:"-"`
AfterStart func() `json:"-"`
BeforeStop func() `json:"-"`
AfterStop func() `json:"-"`
RaftStore raftstore.RaftStore `json:"-"`
ConnPool *util.ConnectPool `json:"-"`
Forbidden bool `json:"-"`
ForbidWriteOpOfProtoVer0 bool `json:"ForbidWriteOpOfProtoVer0"`
}
func (c *MetaPartitionConfig) checkMeta() (err error) {
@ -282,6 +283,8 @@ type MetaPartition interface {
ForceSetMetaPartitionToFininshLoad()
IsForbidden() bool
SetForbidden(status bool)
IsForbidWriteOpOfProtoVer0() bool
SetForbidWriteOpOfProtoVer0(status bool)
IsEnableAuditLog() bool
SetEnableAuditLog(status bool)
UpdateVolumeView(dataView *proto.DataPartitionsView, volumeView *proto.SimpleVolView)
@ -598,6 +601,14 @@ func (mp *metaPartition) SetForbidden(status bool) {
mp.config.Forbidden = status
}
func (mp *metaPartition) IsForbidWriteOpOfProtoVer0() bool {
return mp.config.ForbidWriteOpOfProtoVer0
}
func (mp *metaPartition) SetForbidWriteOpOfProtoVer0(status bool) {
mp.config.ForbidWriteOpOfProtoVer0 = status
}
func (mp *metaPartition) IsEnableAuditLog() bool {
return mp.enableAuditLog
}
@ -978,6 +989,9 @@ func NewMetaPartition(conf *MetaPartitionConfig, manager *metadataManager) MetaP
fmList: newForbiddenMigrationList(proto.ForbiddenMigrationRenewalPeriod),
}
if manager != nil {
mp.config.ForbidWriteOpOfProtoVer0 = manager.isVolForbidWriteOpOfProtoVer0(mp.config.VolName)
}
mp.txProcessor = NewTransactionProcessor(mp)
return mp
}

2
metanode/partition_fsmop_inode.go
View File

@ -508,7 +508,7 @@ func (mp *metaPartition) fsmAppendExtentsWithCheck(ino *Inode, isSplit bool) (st
}
if err := fsmIno.updateStorageClass(storageClass, isCache, isMigration); err != nil {
log.LogErrorf("action[fsmAppendExtentsWithCheck] updateStorageClass inode %v isCache %v isMigration %v, failed %v",
log.LogErrorf("action[fsmAppendExtentsWithCheck] updateStorageClass inode(%v) isCache(%v) isMigration(%v), failed: %v",
ino.Inode, isCache, isMigration, err.Error())
status = proto.OpMismatchStorageClass
return

2
metanode/partition_op_extent.go
View File

@ -122,7 +122,7 @@ func (mp *metaPartition) ExtentAppendWithCheck(req *proto.AppendExtentKeyWithChe
}
if req.StorageClass == proto.StorageClass_Unspecified {
log.LogWarnf("[ExtentAppendWithCheck] mp(%v) inode(%v) reqStorageClass is %v, maybe from lower version client, set it as inoStorageClass(%v)",
log.LogWarnf("[ExtentAppendWithCheck] mp(%v) inode(%v) storageClass(%v), reqStorageClass(%v), maybe from lower version client, set it as inode StorageClass",
mp.config.PartitionId, inoParm.Inode, proto.StorageClassString(inoParm.StorageClass), proto.StorageClassString(req.StorageClass))
req.StorageClass = inoParm.StorageClass
} else if !proto.IsStorageClassReplica(req.StorageClass) {

49
proto/admin_proto.go
View File

@ -551,20 +551,21 @@ type VolumeVerInfo struct {
// ClusterInfo defines the cluster infomation.
type ClusterInfo struct {
Cluster string
Ip string
MetaNodeDeleteBatchCount uint64
MetaNodeDeleteWorkerSleepMs uint64
DataNodeDeleteLimitRate uint64
DataNodeAutoRepairLimitRate uint64
DpMaxRepairErrCnt uint64
DirChildrenNumLimit uint32
EbsAddr string
ServicePath string
ClusterUuid string
ClusterUuidEnable bool
ClusterEnableSnapshot bool
ForbidWriteOpOfProtoVer0 bool
Cluster string
Ip string
MetaNodeDeleteBatchCount uint64
MetaNodeDeleteWorkerSleepMs uint64
DataNodeDeleteLimitRate uint64
DataNodeAutoRepairLimitRate uint64
DpMaxRepairErrCnt uint64
DirChildrenNumLimit uint32
EbsAddr string
ServicePath string
ClusterUuid string
ClusterUuidEnable bool
ClusterEnableSnapshot bool
ForbidWriteOpOfProtoVer0 bool
VolsForbidWriteOpOfProtoVer0 []string
}
// CreateDataPartitionRequest defines the request to create a data partition.
@ -754,12 +755,14 @@ type HeartBeatRequest struct {
UidLimitToMetaNode
QuotaHeartBeatInfos
TxInfos
ForbiddenVols []string
DisableAuditVols []string
DecommissionDisks []string // NOTE: for datanode
VolDpRepairBlockSize map[string]uint64
DpBackupTimeout string
NotifyForbidWriteOpOfProtoVer0 bool
ForbiddenVols []string
DisableAuditVols []string
DecommissionDisks []string // NOTE: for datanode
VolDpRepairBlockSize map[string]uint64
DpBackupTimeout string
NotifyForbidWriteOpOfProtoVer0 bool // whether forbid by node granularity, will notify to nodes
VolsForbidWriteOpOfProtoVer0 []string // whether forbid by volume granularity, will notify to partitions of volume in nodes
}
// DataPartitionReport defines the partition report.
@ -1509,6 +1512,6 @@ func GetStorageClassByMediaType(mediaType uint32) (storageClass uint32) {
return
}
// const ForbiddenMigrationRenewalPeriod = 2 * time.Minute
// TODO:chihe,tangjingyu debug
const ForbiddenMigrationRenewalPeriod = 10 * time.Second
const ForbiddenMigrationRenewalPeriod = 2 * time.Minute
// const ForbiddenMigrationRenewalPeriod = 10 * time.Second // for debug

4
proto/packet.go
View File

@ -413,8 +413,8 @@ func (p *Packet) GetCopy() *Packet {
}
func (p *Packet) String() string {
return fmt.Sprintf("ReqID(%v)Op(%v)PartitionID(%v)ResultCode(%v)ExID(%v)ExtOffset(%v)KernelOff(%v)Type(%v)Seq(%v)Size(%v)",
p.ReqID, p.GetOpMsg(), p.PartitionID, p.GetResultMsg(), p.ExtentID, p.ExtentOffset, p.KernelOffset, p.ExtentType, p.VerSeq, p.Size)
return fmt.Sprintf("ReqID(%v)Op(%v)PartitionID(%v)ResultCode(%v)ExID(%v)ExtOffset(%v)KernelOff(%v)Type(%v)VerSeq(%v)ProtoVer(%v)Size(%v)",
p.ReqID, p.GetOpMsg(), p.PartitionID, p.GetResultMsg(), p.ExtentID, p.ExtentOffset, p.KernelOffset, p.ExtentType, p.VerSeq, p.ProtoVersion, p.Size)
}
// GetStoreType returns the store type.

6
repl/packet.go
View File

@ -136,7 +136,7 @@ func (p *FollowerPacket) identificationErrorResultCode(errLog string, errMsg str
strings.Contains(errLog, ConnIsNullErr) {
p.ResultCode = proto.OpIntraGroupNetErr
log.LogErrorf("action[identificationErrorResultCode] error %v, errmsg %v", errLog, errMsg)
} else if strings.Contains(errMsg, storage.WriteOpOfProtoVerForbidden.Error()) {
} else if strings.Contains(errMsg, storage.ClusterForbidWriteOpOfProtoVer.Error()) {
p.ResultCode = proto.OpWriteOpOfProtoVerForbidden
} else {
log.LogErrorf("action[identificationErrorResultCode] error %v, errmsg %v", errLog, errMsg)
@ -470,7 +470,9 @@ func (p *Packet) identificationErrorResultCode(errLog string, errMsg string) {
p.ResultCode = proto.OpStoreClosed
} else if strings.Contains(errMsg, storage.ReachMaxExtentsCountError.Error()) {
p.ResultCode = proto.OpReachMaxExtentsErr
} else if strings.Contains(errMsg, storage.WriteOpOfProtoVerForbidden.Error()) {
} else if strings.Contains(errMsg, storage.ClusterForbidWriteOpOfProtoVer.Error()) {
p.ResultCode = proto.OpWriteOpOfProtoVerForbidden
} else if strings.Contains(errMsg, storage.VolForbidWriteOpOfProtoVer.Error()) {
p.ResultCode = proto.OpWriteOpOfProtoVerForbidden
} else {
log.LogErrorf("action[identificationErrorResultCode] error %v, errmsg %v", errLog, errMsg)

8
sdk/data/stream/extent_handler.go
View File

@ -123,8 +123,12 @@ type ExtentHandler struct {
// NewExtentHandler returns a new extent handler.
func NewExtentHandler(stream *Streamer, offset int, storeMode int, size int,
storageClass uint32, isMigration bool) *ExtentHandler {
// log.LogDebugf("NewExtentHandler stack(%v)", string(debug.Stack()))
storageClass uint32, isMigration bool,
) *ExtentHandler {
// TODO: for debug
// log.LogDebugf("NewExtentHandler, inode(%v) storageClass(%v), stack:\n%v",
// stream.inode, storageClass, string(debug.Stack()))
eh := &ExtentHandler{
stream: stream,
id: GetExtentHandlerID(),

1
sdk/data/stream/packet.go
View File

@ -148,6 +148,7 @@ func NewCreateExtentPacket(dp *wrapper.DataPartition, inode uint64) *Packet {
p.PartitionID = dp.PartitionID
p.Magic = proto.ProtoMagic
p.ExtentType = proto.NormalExtentType
p.ExtentType |= proto.PacketProtocolVersionFlag
p.Arg = ([]byte)(dp.GetAllAddrs())
p.ArgLen = uint32(len(p.Arg))
p.RemainingFollowers = uint8(len(dp.Hosts) - 1)

1
sdk/data/stream/stream_conn.go
View File

@ -127,6 +127,7 @@ func (sc *StreamConn) getRetryTimeOut() time.Duration {
func (sc *StreamConn) Send(retry *bool, req *Packet, getReply GetReplyFunc) (err error) {
start := time.Now()
retryInterval := StreamSendSleepInterval
req.ExtentType |= proto.PacketProtocolVersionFlag
for i := 0; i < StreamSendMaxRetry; i++ {
err = sc.sendToDataPartition(req, retry, getReply)

13
sdk/data/wrapper/default_random_selector.go
View File

@ -88,7 +88,7 @@ func (s *DefaultRandomSelector) Select(exclude map[string]struct{}, mediaType ui
return dp, nil
}
log.LogErrorf("DefaultRandomSelector: ehID(%v) no writable data partition with %v partitions and exclude(%v)mediaType(%v)",
log.LogErrorf("DefaultRandomSelector: ehID(%v) no writable data partition with %v partitions and exclude(%v) mediaType(%v)",
ehID, len(partitions), exclude, proto.MediaTypeString(mediaType))
return nil, fmt.Errorf("en(%v) no writable data partition", ehID)
}
@ -171,19 +171,20 @@ func (s *DefaultRandomSelector) getRandomDataPartition(partitions []*DataPartiti
index := rand.Intn(length)
dp = partitions[index]
if !isExcluded(dp, exclude) && dp.MediaType == mediaType {
log.LogDebugf("DefaultRandomSelector: eh(%v) select dp[%v] address[%p], index %v", ehID, dp, dp, index)
log.LogDebugf("DefaultRandomSelector: eh(%v) select dp(%v) index(%v)", ehID, dp.PartitionID, index)
return dp
}
log.LogDebugf("DefaultRandomSelector: eh(%v)first random partition was excluded, get partition from others", ehID)
log.LogDebugf("DefaultRandomSelector: eh(%v) first random partition(%v) MediaType(%v) was excluded, get partition from others",
ehID, dp.PartitionID, proto.MediaTypeString(dp.MediaType))
var currIndex int
for i := 0; i < length; i++ {
currIndex = (index + i) % length
dp = partitions[currIndex]
if !isExcluded(dp, exclude) && dp.MediaType == mediaType {
log.LogDebugf("DefaultRandomSelector: eh(%v) select dp[%v], index %v", ehID, partitions[currIndex], currIndex)
return partitions[currIndex]
log.LogDebugf("DefaultRandomSelector: eh(%v) select dp(%v) MediaType(%v), index %v",
ehID, dp.PartitionID, proto.MediaTypeString(dp.MediaType), currIndex)
return dp
}
}
return nil

4
sdk/master/api_admin.go
View File

@ -301,6 +301,7 @@ func (api *AdminAPI) UpdateVolume(
request.addParam("accessTimeValidInterval", strconv.FormatInt(vv.AccessTimeInterval, 10))
request.addParam("enablePersistAccessTime", strconv.FormatBool(vv.EnablePersistAccessTime))
request.addParam("volStorageClass", strconv.FormatUint(uint64(vv.VolStorageClass), 10))
request.addParam("forbidWriteOpOfProtoVersion0", strconv.FormatBool(vv.ForbidWriteOpOfProtoVer0))
if txMask != "" {
request.addParam("enableTxMask", txMask)
@ -545,7 +546,8 @@ func (api *AdminAPI) SetMasterVolDeletionDelayTime(volDeletionDelayTimeHour int)
func (api *AdminAPI) SetClusterParas(batchCount, markDeleteRate, deleteWorkerSleepMs, autoRepairRate, loadFactor, maxDpCntLimit, maxMpCntLimit, clientIDKey string,
enableAutoDecommissionDisk string, autoDecommissionDiskInterval string,
enableAutoDpMetaRepair string, autoDpMetaRepairParallelCnt string,
dpRepairTimeout string, dpTimeout string, dpBackupTimeout string, decommissionDpLimit, decommissionDiskLimit, forbidWriteOpOfProtoVersion0 string,
dpRepairTimeout string, dpTimeout string, dpBackupTimeout string,
decommissionDpLimit, decommissionDiskLimit, forbidWriteOpOfProtoVersion0 string,
) (err error) {
request := newRequest(get, proto.AdminSetNodeInfo).Header(api.h)
request.addParam("batchCount", batchCount)

4
sdk/meta/conn.go
View File

@ -75,6 +75,8 @@ func (mw *MetaWrapper) sendToMetaPartition(mp *MetaPartition, req *proto.Packet)
delta := (sendTimeLimit*2/SendRetryLimit - SendRetryInterval*2) / SendRetryLimit // ms
log.LogDebugf("mw.metaSendTimeout: %v s, sendTimeLimit: %v ms, delta: %v ms, req %v", mw.metaSendTimeout, sendTimeLimit, delta, req)
req.ExtentType |= proto.PacketProtocolVersionFlag
errs := make(map[int]error, len(mp.Members))
var j int
@ -160,6 +162,8 @@ out:
}
func (mc *MetaConn) send(req *proto.Packet) (resp *proto.Packet, err error) {
req.ExtentType |= proto.PacketProtocolVersionFlag
err = req.WriteToConn(mc.conn)
if err != nil {
return nil, errors.Trace(err, "Failed to write to conn, req(%v)", req)

6
sdk/meta/operation.go
View File

@ -16,7 +16,6 @@ package meta
import (
"fmt"
"runtime/debug"
"strconv"
"sync"
"syscall"
@ -1315,8 +1314,9 @@ func (mw *MetaWrapper) getExtents(mp *MetaPartition, inode uint64, isCache bool,
defer func() {
metric.SetWithLabels(err, map[string]string{exporter.Vol: mw.volname})
}()
// TODO:chi,tangjingyu: delete this line when the bug fixed
log.LogDebugf("getExtents req: id(%v) data(%v) stack[%v]", packet.GetReqID(), req, string(debug.Stack()))
// for debug
// log.LogDebugf("getExtents req: id(%v) data(%v) stack[%v]", packet.GetReqID(), req, string(debug.Stack()))
packet, err = mw.sendToMetaPartition(mp, packet)
if err != nil {

3
storage/error.go
View File

@ -41,7 +41,8 @@ var (
SnapshotNeedNewExtentError = errors.New("snapshot need new extent error")
NoDiskReadRepairExtentTokenError = errors.New("no disk read repair extent token")
ReachMaxExtentsCountError = errors.New("reached max extents count")
WriteOpOfProtoVerForbidden = errors.New("forbid write operate of packet protocol version")
ClusterForbidWriteOpOfProtoVer = errors.New("cluster forbid write operate of packet protocol version")
VolForbidWriteOpOfProtoVer = errors.New("vol forbid write operate of packet protocol version")
)
func newParameterError(format string, a ...interface{}) error {