From 71dee686ef51616360d7451f41ac450b823cfa3d Mon Sep 17 00:00:00 2001 From: zhumingze Date: Tue, 27 May 2025 19:53:25 +0800 Subject: [PATCH] feat(data): limitIO strategy for separating repair io from business io. #1000095860 Signed-off-by: zhumingze --- cmd/common/args.go | 2 + datanode/data_partition_repair.go | 64 ++++--- datanode/data_partition_repair_test.go | 127 +++++++++++++- datanode/disk.go | 125 +++++++++++++- datanode/partition.go | 8 +- datanode/partition_op_by_raft.go | 5 +- datanode/server.go | 60 +++++-- datanode/server_handler.go | 63 +++++-- datanode/storage/extent_store.go | 1 - datanode/wrap_operator.go | 60 +++---- datanode/wrap_operator_test.go | 227 +++++++++++++++++++------ master/api_service.go | 21 ++- master/api_service_test.go | 2 +- master/cluster.go | 6 +- master/data_node.go | 4 +- master/data_partition_map.go | 5 +- master/gapi_cluster.go | 2 +- proto/admin_proto.go | 22 ++- 18 files changed, 635 insertions(+), 169 deletions(-) diff --git a/cmd/common/args.go b/cmd/common/args.go index bf5761bcb..1fad60cda 100644 --- a/cmd/common/args.go +++ b/cmd/common/args.go @@ -49,3 +49,5 @@ func (s *String) Disk() *Argument { return s.Key("disk") } func (s *String) DiskPath() *Argument { return s.Key("diskPath") } func (s *String) Addr() *Argument { return s.Key("addr") } func (s *String) ZoneName() *Argument { return s.Key("zoneName") } + +func (s *String) QosType() *Argument { return s.Key("qosType") } diff --git a/datanode/data_partition_repair.go b/datanode/data_partition_repair.go index 69c710d7f..629d1f348 100644 --- a/datanode/data_partition_repair.go +++ b/datanode/data_partition_repair.go @@ -292,9 +292,10 @@ func (dp *DataPartition) DoRepair(repairTasks []*DataPartitionRepairTask) { continue } - dp.disk.allocCheckLimit(proto.IopsWriteType, 1) - - err := store.Create(extentInfo.FileID) + var err error + dp.disk.diskLimit(OpAsyncWrite, 0, func() { + err = store.Create(extentInfo.FileID) + }) if err != nil { log.LogWarnf("DoRepair dp %v extent %v failed, err:%v", dp.partitionID, extentInfo.FileID, err.Error()) @@ -573,11 +574,16 @@ func (dp *DataPartition) ExtentWithHoleRepairRead(request repl.PacketInterface, reply.SetData(make([]byte, currReadSize)) } reply.SetExtentOffset(offset) - crc, err = dp.extentStore.Read(reply.GetExtentID(), offset, int64(currReadSize), reply.GetData(), false, request.GetOpcode() == proto.OpBackupRead) + + dp.disk.diskLimit(OpAsyncRead, currReadSize, func() { + crc, err = dp.extentStore.Read(reply.GetExtentID(), offset, int64(currReadSize), reply.GetData(), false, request.GetOpcode() == proto.OpBackupRead) + }) + dp.checkIsDiskError(err, ReadFlag) if err != nil { return } + reply.SetCRC(crc) reply.SetSize(currReadSize) reply.SetResultCode(proto.OpOk) @@ -606,6 +612,8 @@ func (dp *DataPartition) NormalExtentRepairRead(p repl.PacketInterface, connect var ( metricPartitionIOLabels map[string]string partitionIOMetric, tpObject *exporter.TimePointCount + crc uint32 + opType string ) shallDegrade := p.ShallDegrade() if !shallDegrade { @@ -643,16 +651,16 @@ func (dp *DataPartition) NormalExtentRepairRead(p repl.PacketInterface, connect p.SetSize(currReadSize) p.SetExtentOffset(offset) - dp.Disk().allocCheckLimit(proto.IopsReadType, 1) - dp.Disk().allocCheckLimit(proto.FlowReadType, currReadSize) - - if rs := dp.disk.limitRead.Run(int(currReadSize), false, func() { - var crc uint32 + if isRepairRead { + opType = OpAsyncRead + } else { + opType = OpRead + } + dp.disk.diskLimit(opType, currReadSize, func() { crc, err = store.Read(reply.GetExtentID(), offset, int64(currReadSize), reply.GetData(), isRepairRead, p.GetOpcode() == proto.OpBackupRead) reply.SetCRC(crc) - }); err == nil && rs != nil { - err = rs - } + }) + if !shallDegrade && metrics != nil { metrics.MetricIOBytes.AddWithLabels(int64(p.GetSize()), metricPartitionIOLabels) partitionIOMetric.SetWithLabels(err, metricPartitionIOLabels) @@ -890,7 +898,9 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *RepairExtentInfo, log.LogDebugf("streamRepairExtent dp[%v] extent[%v] localExtentInfo[%v] remote info(remoteAvaliSize[%v],isEmptyResponse[%v],currRecoverySize[%v] currFixOffset[%v]", dp.partitionID, localExtentInfo, remoteExtentInfo, remoteAvaliSize, isEmptyResponse, currRecoverySize, currFixOffset) if storage.IsTinyExtent(localExtentInfo.FileID) { - err = store.TinyExtentRecover(uint64(localExtentInfo.FileID), int64(currFixOffset), int64(currRecoverySize), reply.GetData(), reply.GetCRC(), isEmptyResponse) + dp.disk.diskLimit(OpAsyncWrite, uint32(currRecoverySize), func() { + err = store.TinyExtentRecover(uint64(localExtentInfo.FileID), int64(currFixOffset), int64(currRecoverySize), reply.GetData(), reply.GetCRC(), isEmptyResponse) + }) if hasRecoverySize+currRecoverySize >= remoteAvaliSize { log.LogInfof("streamRepairTinyExtent(%v) recover fininsh,remoteAvaliSize(%v) "+ "hasRecoverySize(%v) currRecoverySize(%v)", dp.applyRepairKey(int(localExtentInfo.FileID)), @@ -899,19 +909,21 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *RepairExtentInfo, } } else { log.LogDebugf("streamRepairExtent reply size %v, currFixoffset %v, reply %v ", reply.GetSize(), currFixOffset, reply) - param := &storage.WriteParam{ - ExtentID: uint64(localExtentInfo.FileID), - Offset: int64(currFixOffset), - Size: int64(reply.GetSize()), - Data: reply.GetData(), - Crc: reply.GetCRC(), - WriteType: wType, - IsSync: BufferWrite, - IsHole: isEmptyResponse, - IsRepair: true, - IsBackupWrite: request.GetOpcode() == proto.OpBackupWrite, - } - _, err = store.Write(param) + dp.disk.diskLimit(OpAsyncWrite, uint32(reply.GetSize()), func() { + param := &storage.WriteParam{ + ExtentID: uint64(localExtentInfo.FileID), + Offset: int64(currFixOffset), + Size: int64(reply.GetSize()), + Data: reply.GetData(), + Crc: reply.GetCRC(), + WriteType: wType, + IsSync: BufferWrite, + IsHole: isEmptyResponse, + IsRepair: true, + IsBackupWrite: request.GetOpcode() == proto.OpBackupWrite, + } + _, err = store.Write(param) + }) } // log.LogDebugf("streamRepairExtent reply size %v, currFixoffset %v, reply %v err %v", reply.Size, currFixOffset, reply, err) // write to the local extent file diff --git a/datanode/data_partition_repair_test.go b/datanode/data_partition_repair_test.go index cd0d84553..45499942e 100644 --- a/datanode/data_partition_repair_test.go +++ b/datanode/data_partition_repair_test.go @@ -399,7 +399,11 @@ func mockInitWorker(t *testing.T, role string) *repairWorker { require.NoError(t, err) worker.dp = mockMakeDp(path) - spaceManager := NewSpaceManager(worker.dp.dataNode) + // spaceManager := NewSpaceManager(worker.dp.dataNode) + spaceManager := &SpaceManager{ + dataNode: worker.dp.dataNode, + partitions: make(map[uint64]*DataPartition), + } worker.dp.disk, err = NewDisk("/tmp", 200, 2000, 10, spaceManager, false) require.NoError(t, err) spaceManager.partitions[worker.dp.partitionID] = worker.dp @@ -586,3 +590,124 @@ func TestExtentRepair_512KB(t *testing.T) { data, crc = genDataAndGetCrc("snapshot", util.BlockSize) testDoSnapshotRepair(t, normalId, data, crc, false) } + +func TestExtentRepairWithIOLimit(t *testing.T) { + proto.InitBufferPool(int64(32768)) + t.Logf("TestExtentRepairWithIOLimit initWorker") + + const repairRateLimit = 1 * util.MB + + // normalIds := []uint64{1025, 1026, 1027, 1028, 1029, 1030, 1031, 1032, 1033, 1034} + normalIds := make([]uint64, 10) + for i := 0; i < 10; i++ { + normalIds[i] = uint64(1025 + i) + } + data, crc := genDataAndGetCrc("normal", util.BlockSize) + + sendWorker = mockInitWorker(t, "sender") + recvWorker = mockInitWorker(t, "receiver") + sendWorker.dstWorker = recvWorker + recvWorker.dstWorker = sendWorker + + sendWorker.dp.disk.limitAsyncRead = util.NewIOLimiter(repairRateLimit, 0) + recvWorker.dp.disk.limitAsyncWrite = util.NewIOLimiter(repairRateLimit, 0) + + for _, id := range normalIds { + err := sendWorker.dp.extentStore.Create(id) + require.NoError(t, err) + err = recvWorker.dp.extentStore.Create(id) + require.NoError(t, err) + extentStoreNormalRwTest(t, sendWorker.dp.extentStore, id, crc, data) + } + + defer func() { + os.RemoveAll(filepath.Dir(sendWorker.dp.path)) + os.RemoveAll(filepath.Dir(recvWorker.dp.path)) + }() + testConcurrentRepairs(t, normalIds, repairRateLimit) +} + +func testConcurrentRepairs(t *testing.T, normalIds []uint64, repairRateLimit int64) { + var wg sync.WaitGroup + exitCh := make(chan struct{}) + startTime := time.Now() + + defer func() { + totalDuration := time.Since(startTime) + t.Logf("totalDuration: %v", totalDuration) + expectedMinDuration := time.Duration((len(normalIds)*util.BlockSize)/(int(repairRateLimit))) * time.Second + t.Logf("expectedMinDuration: %v", expectedMinDuration) + assert.True(t, totalDuration > expectedMinDuration, + fmt.Sprintf("actural cost time %v should > expected cost time %v", totalDuration, expectedMinDuration)) + }() + + wg.Add(1) + go func() { + defer wg.Done() + senderRepairWorkerWithConcurrent(t, exitCh) + }() + + wg.Add(1) + go func() { + defer wg.Done() + recvRepairWorkerWithConcurrent(t, normalIds, exitCh) + }() + + wg.Wait() + close(exitCh) +} + +func senderRepairWorkerWithConcurrent(t *testing.T, exitCh chan struct{}) { + con := new(net.TCPConn) + var wg sync.WaitGroup + sem := make(chan struct{}, 10) + + for { + select { + case pr := <-sendWorker.packChannel: + if pr.GetOpcode() == proto.OpExtentRepairRead { + sem <- struct{}{} + wg.Add(1) + go func(p repl.PacketInterface) { + defer func() { + <-sem + wg.Done() + }() + start := time.Now() + t.Logf("Sender: limitAsyncRead %v", sendWorker.dp.disk.limitAsyncRead.Status(false)) + sendWorker.dp.NormalExtentRepairRead(pr, con, true, nil, sendNewNormalReadResponsePacket) + elapsed := time.Since(start) + t.Logf("request %d read cost time: %v", pr.GetExtentID(), elapsed) + }(pr) + } + case <-exitCh: + wg.Wait() + return + } + } +} + +func recvRepairWorkerWithConcurrent(t *testing.T, ids []uint64, exitCh chan struct{}) { + var wg sync.WaitGroup + for _, id := range ids { + wg.Add(1) + go func(eid uint64) { + defer wg.Done() + ei, err := sendWorker.dp.extentStore.Watermark(eid) + require.NoError(t, err) + t.Logf("Recver: limitAsyncWrite %v", recvWorker.dp.disk.limitAsyncWrite.Status(false)) + repairExtent := &RepairExtentInfo{ExtentInfo: *ei} + start := time.Now() + err = recvWorker.dp.streamRepairExtent(repairExtent, + reciverMakeTinyPacket, + reciverMakeNormalPacket, + reciverMakeExtentWithHoleRepairReadPacket, + reciverNewPacket) + assert.NoError(t, err) + elapsed := time.Since(start) + t.Logf("request %d write cost time: %v", eid, elapsed) + }(id) + } + wg.Wait() + exitCh <- struct{}{} +} diff --git a/datanode/disk.go b/datanode/disk.go index 618a8536b..0cab99228 100644 --- a/datanode/disk.go +++ b/datanode/disk.go @@ -61,6 +61,14 @@ const ( DecommissionDiskMark = "decommissionDiskMark" ) +const ( + OpRead = "read" + OpAsyncRead = "asyncRead" + OpWrite = "write" + OpAsyncWrite = "asyncWrite" + OpDelete = "delete" +) + // Disk represents the structure of the disk type Disk struct { sync.RWMutex @@ -86,9 +94,12 @@ type Disk struct { space *SpaceManager dataNode *DataNode - limitFactor map[uint32]*rate.Limiter - limitRead *util.IoLimiter - limitWrite *util.IoLimiter + limitFactor map[uint32]*rate.Limiter + limitRead *util.IoLimiter + limitWrite *util.IoLimiter + limitAsyncRead *util.IoLimiter + limitAsyncWrite *util.IoLimiter + limitDelete *util.IoLimiter // diskPartition info diskPartition *disk.PartitionStat @@ -157,8 +168,16 @@ func NewDisk(path string, reservedSpace, diskRdonlySpace uint64, maxErrCnt int, d.limitFactor[proto.FlowWriteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxFLowLimit), proto.QosDefaultBurst) d.limitFactor[proto.IopsReadType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) d.limitFactor[proto.IopsWriteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) + d.limitFactor[proto.FlowAsyncReadType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxFLowLimit), proto.QosDefaultBurst) + d.limitFactor[proto.FlowAsyncWriteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxFLowLimit), proto.QosDefaultBurst) + d.limitFactor[proto.IopsAsyncReadType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) + d.limitFactor[proto.IopsAsyncWriteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) + d.limitFactor[proto.IopsDeleteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultMarkDeleteLimitBurst) d.limitRead = util.NewIOLimiter(space.dataNode.diskReadFlow, space.dataNode.diskReadIocc) d.limitWrite = util.NewIOLimiter(space.dataNode.diskWriteFlow, space.dataNode.diskWriteIocc) + d.limitAsyncRead = util.NewIOLimiter(space.dataNode.diskAsyncReadFlow, space.dataNode.diskAsyncReadIocc) + d.limitAsyncWrite = util.NewIOLimiter(space.dataNode.diskAsyncWriteFlow, space.dataNode.diskAsyncWriteIocc) + d.limitDelete = util.NewIOLimiter(space.dataNode.diskDeleteFlow, space.dataNode.diskDeleteIocc) err = d.initDecommissionStatus() if err != nil { @@ -256,7 +275,7 @@ func (d *Disk) isBrokenDisk() (ok bool) { func (d *Disk) updateQosLimiter() { if d.isBrokenDisk() || d.isLost { - log.LogInfof("[updateQosLimiter] disk(%v) is broken", d.Path) + log.LogInfof("[updateQosLimiter] disk(%v) is broken or lost", d.Path) return } if d.dataNode.diskReadFlow > 0 { @@ -271,16 +290,38 @@ func (d *Disk) updateQosLimiter() { if d.dataNode.diskWriteIops > 0 { d.limitFactor[proto.IopsWriteType].SetLimit(rate.Limit(d.dataNode.diskWriteIops)) } - for i := proto.IopsReadType; i < proto.FlowWriteType; i++ { + if d.dataNode.diskAsyncReadFlow > 0 { + d.limitFactor[proto.FlowAsyncReadType].SetLimit(rate.Limit(d.dataNode.diskAsyncReadFlow)) + } + if d.dataNode.diskAsyncWriteFlow > 0 { + d.limitFactor[proto.FlowAsyncWriteType].SetLimit(rate.Limit(d.dataNode.diskAsyncWriteFlow)) + } + if d.dataNode.diskAsyncReadIops > 0 { + d.limitFactor[proto.IopsAsyncReadType].SetLimit(rate.Limit(d.dataNode.diskAsyncReadIops)) + } + if d.dataNode.diskAsyncWriteIops > 0 { + d.limitFactor[proto.IopsAsyncWriteType].SetLimit(rate.Limit(d.dataNode.diskAsyncWriteIops)) + } + if d.dataNode.diskDeleteIops > 0 { + d.limitFactor[proto.IopsDeleteType].SetLimit(rate.Limit(d.dataNode.diskDeleteIops)) + } + for i := proto.IopsReadType; i < proto.FlowDeleteType; i++ { log.LogInfof("action[updateQosLimiter] type %v limit %v", proto.QosTypeString(i), d.limitFactor[i].Limit()) } - log.LogWarnf("action[updateQosLimiter] read(iocc:%d iops:%d flow:%d) write(iocc:%d iops:%d flow:%d)", - d.dataNode.diskReadIocc, d.dataNode.diskReadIops, d.dataNode.diskReadFlow, - d.dataNode.diskWriteIocc, d.dataNode.diskWriteIops, d.dataNode.diskWriteFlow) + log.LogWarnf("action[updateQosLimiter] read(iocc:normal %d async %d, iops:%d async %d, flow:normal %d async %d) write(iocc:%d async %d, iops:%d async %d, flow:%d async %d) delete(iocc:%d, flow:%d, iops:%d)", + d.dataNode.diskReadIocc, d.dataNode.diskAsyncReadIocc, d.dataNode.diskReadIops, d.dataNode.diskAsyncReadIops, d.dataNode.diskReadFlow, d.dataNode.diskAsyncReadFlow, + d.dataNode.diskWriteIocc, d.dataNode.diskAsyncWriteIocc, d.dataNode.diskWriteIops, d.dataNode.diskAsyncWriteIops, d.dataNode.diskWriteFlow, d.dataNode.diskAsyncWriteFlow, + d.dataNode.diskDeleteIocc, d.dataNode.diskDeleteFlow, d.dataNode.diskDeleteIops) d.limitRead.ResetIO(d.dataNode.diskReadIocc, 0) d.limitRead.ResetFlow(d.dataNode.diskReadFlow) d.limitWrite.ResetIO(d.dataNode.diskWriteIocc, d.dataNode.diskWQueFactor) d.limitWrite.ResetFlow(d.dataNode.diskWriteFlow) + d.limitAsyncRead.ResetIO(d.dataNode.diskAsyncReadIocc, 0) + d.limitAsyncRead.ResetFlow(d.dataNode.diskAsyncReadFlow) + d.limitAsyncWrite.ResetIO(d.dataNode.diskAsyncWriteIocc, 0) + d.limitAsyncWrite.ResetFlow(d.dataNode.diskAsyncWriteFlow) + d.limitDelete.ResetIO(d.dataNode.diskDeleteIocc, 0) + d.limitDelete.ResetFlow(d.dataNode.diskDeleteFlow) } func (d *Disk) allocCheckLimit(factorType uint32, used uint32) error { @@ -293,6 +334,74 @@ func (d *Disk) allocCheckLimit(factorType uint32, used uint32) error { return nil } +func (d *Disk) allocCheckAsyncLimit(factorType uint32, used uint32) error { + if !d.dataNode.diskAsyncQosEnable { + return nil + } + + ctx := context.Background() + d.limitFactor[factorType].WaitN(ctx, int(used)) + return nil +} + +func (d *Disk) getLimitIoConfig(ioType string) ( + flowType, iopsType uint32, + allocCheckFunc func(factorType uint32, used uint32) error, + limiter *util.IoLimiter, + allowHang bool, +) { + switch ioType { + case OpRead: + return proto.FlowReadType, proto.IopsReadType, d.allocCheckLimit, d.limitRead, false + case OpAsyncRead: + return proto.FlowAsyncReadType, proto.IopsAsyncReadType, d.allocCheckAsyncLimit, d.limitAsyncRead, true + case OpWrite: + return proto.FlowWriteType, proto.IopsWriteType, d.allocCheckLimit, d.limitWrite, true + case OpAsyncWrite: + return proto.FlowAsyncWriteType, proto.IopsAsyncWriteType, d.allocCheckAsyncLimit, d.limitAsyncWrite, true + case OpDelete: + return proto.FlowDeleteType, proto.IopsDeleteType, d.allocCheckAsyncLimit, d.limitDelete, true + default: + panic("unknown ioType: " + ioType) + } +} + +func (d *Disk) diskLimit( + ioType string, + operationSize uint32, + operationFunc func(), +) { + flowType, iopsType, allocCheckFunc, limiter, allowHang := d.getLimitIoConfig(ioType) + + if operationSize > 0 { + allocCheckFunc(flowType, operationSize) + } + allocCheckFunc(iopsType, 1) + + limiter.Run(int(operationSize), allowHang, func() { + operationFunc() + }) +} + +func (d *Disk) tryDiskLimit( + ioType string, + operationSize uint32, + operationFunc func(), +) bool { + flowType, iopsType, allocCheckFunc, limiter, _ := d.getLimitIoConfig(ioType) + + if operationSize > 0 { + allocCheckFunc(flowType, operationSize) + } + allocCheckFunc(iopsType, 1) + + writable := limiter.TryRun(int(operationSize), func() { + operationFunc() + }) + + return writable +} + // PartitionCount returns the number of partitions in the partition map. func (d *Disk) PartitionCount() int { d.RLock() diff --git a/datanode/partition.go b/datanode/partition.go index 91e60284b..79375ec39 100644 --- a/datanode/partition.go +++ b/datanode/partition.go @@ -1142,9 +1142,10 @@ func (dp *DataPartition) DoExtentStoreRepair(repairTask *DataPartitionRepairTask continue } - dp.disk.allocCheckLimit(proto.IopsWriteType, 1) - - err := store.Create(uint64(extentInfo.FileID)) + var err error + dp.disk.diskLimit(OpAsyncWrite, 0, func() { + err = store.Create(extentInfo.FileID) + }) if err != nil { log.LogWarnf("DoExtentStoreRepair dp %v extent %v failed, err:%v", dp.partitionID, extentInfo.FileID, err.Error()) @@ -1306,7 +1307,6 @@ func (dp *DataPartition) doStreamFixTinyDeleteRecord(repairTask *DataPartitionRe continue } DeleteLimiterWait() - dp.disk.allocCheckLimit(proto.IopsWriteType, 1) // log.LogInfof("doStreamFixTinyDeleteRecord Delete PartitionID(%v)_Extent(%v)_Offset(%v)_Size(%v)", dp.partitionID, extentID, offset, size) // store.MarkDelete(extentID, int64(offset), int64(size)) store.RecordTinyDelete(extentID, int64(offset), int64(size)) diff --git a/datanode/partition_op_by_raft.go b/datanode/partition_op_by_raft.go index bdd7045cc..48c556eb5 100644 --- a/datanode/partition_op_by_raft.go +++ b/datanode/partition_op_by_raft.go @@ -263,9 +263,6 @@ func (dp *DataPartition) ApplyRandomWrite(command []byte, raftApplyID uint64) (r raftApplyID, dp.partitionID, opItem.extentID, opItem.offset, opItem.size) for i := 0; i < 20; i++ { - dp.disk.allocCheckLimit(proto.FlowWriteType, uint32(opItem.size)) - dp.disk.allocCheckLimit(proto.IopsWriteType, 1) - var syncWrite bool writeType := storage.RandomWriteType if opItem.opcode == proto.OpRandomWrite || opItem.opcode == proto.OpSyncRandomWrite { @@ -284,7 +281,7 @@ func (dp *DataPartition) ApplyRandomWrite(command []byte, raftApplyID uint64) (r syncWrite = true } - dp.disk.limitWrite.Run(int(opItem.size), true, func() { + dp.disk.diskLimit(OpWrite, uint32(opItem.size), func() { param := &storage.WriteParam{ ExtentID: uint64(opItem.extentID), Offset: int64(opItem.offset), diff --git a/datanode/server.go b/datanode/server.go index 18db70b03..60f123d2c 100644 --- a/datanode/server.go +++ b/datanode/server.go @@ -114,14 +114,24 @@ const ( ConfigKeySmuxTotalStream = "sumxTotalStream" // int // rate limit control enable - ConfigDiskQosEnable = "diskQosEnable" // bool - ConfigDiskReadIocc = "diskReadIocc" // int - ConfigDiskReadIops = "diskReadIops" // int - ConfigDiskReadFlow = "diskReadFlow" // int - ConfigDiskWriteIocc = "diskWriteIocc" // int - ConfigDiskWriteIops = "diskWriteIops" // int - ConfigDiskWriteFlow = "diskWriteFlow" // int - ConfigDiskWQueFactor = "diskWQueFactor" // int + ConfigDiskQosEnable = "diskQosEnable" // bool + ConfigDiskAsyncQosEnable = "diskAsyncQosEnable" // bool + ConfigDiskReadIocc = "diskReadIocc" // int + ConfigDiskReadIops = "diskReadIops" // int + ConfigDiskReadFlow = "diskReadFlow" // int + ConfigDiskWriteIocc = "diskWriteIocc" // int + ConfigDiskWriteIops = "diskWriteIops" // int + ConfigDiskWriteFlow = "diskWriteFlow" // int + ConfigDiskWQueFactor = "diskWQueFactor" // int + ConfigDiskAsyncReadIocc = "diskAsyncReadIocc" // int + ConfigDiskAsyncReadIops = "diskAsyncReadIops" // int + ConfigDiskAsyncReadFlow = "diskAsyncReadFlow" // int + ConfigDiskAsyncWriteIocc = "diskAsyncWriteIocc" // int + ConfigDiskAsyncWriteIops = "diskAsyncWriteIops" // int + ConfigDiskAsyncWriteFlow = "diskAsyncWriteFlow" // int + ConfigDiskDeleteIocc = "diskDeleteIocc" // int + ConfigDiskDeleteIops = "diskDeleteIops" // int + ConfigDiskDeleteFlow = "diskDeleteFlow" // int // load/stop dp limit ConfigDiskCurrentLoadDpLimit = "diskCurrentLoadDpLimit" @@ -191,12 +201,22 @@ type DataNode struct { diskQosEnable bool diskQosEnableFromMaster bool + diskAsyncQosEnable bool diskReadIocc int diskReadIops int diskReadFlow int diskWriteIocc int diskWriteIops int diskWriteFlow int + diskAsyncReadIocc int + diskAsyncReadIops int + diskAsyncReadFlow int + diskAsyncWriteIocc int + diskAsyncWriteIops int + diskAsyncWriteFlow int + diskDeleteIocc int + diskDeleteIops int + diskDeleteFlow int diskWQueFactor int dpMaxRepairErrCnt uint64 clusterUuid string @@ -490,14 +510,25 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) { func (s *DataNode) initQosLimit(cfg *config.Config) { dn := s.space.dataNode dn.diskQosEnable = cfg.GetBoolWithDefault(ConfigDiskQosEnable, true) + dn.diskAsyncQosEnable = cfg.GetBoolWithDefault(ConfigDiskAsyncQosEnable, true) dn.diskReadIocc = cfg.GetInt(ConfigDiskReadIocc) dn.diskReadIops = cfg.GetInt(ConfigDiskReadIops) dn.diskReadFlow = cfg.GetInt(ConfigDiskReadFlow) dn.diskWriteIocc = cfg.GetInt(ConfigDiskWriteIocc) dn.diskWriteIops = cfg.GetInt(ConfigDiskWriteIops) dn.diskWriteFlow = cfg.GetInt(ConfigDiskWriteFlow) - log.LogWarnf("action[initQosLimit] set qos [%v], read(iocc:%d iops:%d flow:%d) write(iocc:%d iops:%d flow:%d)", - dn.diskQosEnable, dn.diskReadIocc, dn.diskReadIops, dn.diskReadFlow, dn.diskWriteIocc, dn.diskWriteIops, dn.diskWriteFlow) + dn.diskAsyncReadIocc = cfg.GetInt(ConfigDiskAsyncReadIocc) + dn.diskAsyncReadIops = cfg.GetInt(ConfigDiskAsyncReadIops) + dn.diskAsyncReadFlow = cfg.GetInt(ConfigDiskAsyncReadFlow) + dn.diskAsyncWriteIocc = cfg.GetInt(ConfigDiskAsyncWriteIocc) + dn.diskAsyncWriteIops = cfg.GetInt(ConfigDiskAsyncWriteIops) + dn.diskAsyncWriteFlow = cfg.GetInt(ConfigDiskAsyncWriteFlow) + dn.diskDeleteIocc = cfg.GetInt(ConfigDiskDeleteIocc) + dn.diskDeleteFlow = cfg.GetInt(ConfigDiskDeleteFlow) + dn.diskDeleteIops = cfg.GetInt(ConfigDiskDeleteIops) + log.LogWarnf("action[initQosLimit] set qos [normal %v async %v], rWriteiocc:normal %d async %d, iops:%d async %d, flow:normal %d async %d) write(iocc:%d async %d,iops:%d async %d, flow:%d async %d) delete(iocc:%d flow:%d iops: %d)", + dn.diskQosEnable, dn.diskAsyncQosEnable, dn.diskReadIocc, dn.diskAsyncReadIocc, dn.diskReadIops, dn.diskAsyncReadIops, dn.diskReadFlow, dn.diskAsyncReadFlow, dn.diskWriteIocc, dn.diskAsyncWriteIocc, + dn.diskWriteIops, dn.diskAsyncWriteIops, dn.diskWriteFlow, dn.diskAsyncWriteFlow, dn.diskDeleteIocc, dn.diskDeleteFlow, dn.diskDeleteIops) } func (s *DataNode) updateQosLimit() { @@ -1257,3 +1288,12 @@ func IsDiskErr(errMsg string) bool { return strings.Contains(errMsg, syscall.EIO.Error()) || strings.Contains(errMsg, syscall.EROFS.Error()) } + +func (s *DataNode) IopsStatus() (status proto.IopsStatus) { + status.ReadIops = s.diskReadIops + status.WriteIops = s.diskWriteIops + status.AsyncReadIops = s.diskAsyncReadIops + status.AsyncWriteIops = s.diskAsyncWriteIops + status.DeleteIops = s.diskDeleteIops + return +} diff --git a/datanode/server_handler.go b/datanode/server_handler.go index 0b3a9b2af..e21a6accf 100644 --- a/datanode/server_handler.go +++ b/datanode/server_handler.go @@ -308,12 +308,22 @@ func (s *DataNode) getNormalDeleted(w http.ResponseWriter, r *http.Request) { func (s *DataNode) setQosEnable() func(http.ResponseWriter, *http.Request) { return func(w http.ResponseWriter, r *http.Request) { var enable common.Bool - if err := parseArgs(r, enable.Enable()); err != nil { + var qosType common.String + if err := parseArgs(r, enable.Enable(), qosType.QosType()); err != nil { s.buildFailureResp(w, http.StatusBadRequest, err.Error()) return } - s.diskQosEnable = enable.V - s.buildSuccessResp(w, "success") + switch qosType.V { + case "normal": + s.diskQosEnable = enable.V + case "async": + s.diskAsyncQosEnable = enable.V + default: + err := fmt.Errorf("qos type %v not exist", qosType.V) + s.buildFailureResp(w, http.StatusBadRequest, err.Error()) + return + } + s.buildSuccessResp(w, fmt.Sprintf("Successfully set %v QosEnable to %v", qosType.V, enable.V)) } } @@ -334,13 +344,22 @@ func (s *DataNode) setDiskQos(w http.ResponseWriter, r *http.Request) { updated := false for key, pVal := range map[string]*int{ - ConfigDiskReadIocc: &s.diskReadIocc, - ConfigDiskReadIops: &s.diskReadIops, - ConfigDiskReadFlow: &s.diskReadFlow, - ConfigDiskWriteIocc: &s.diskWriteIocc, - ConfigDiskWriteIops: &s.diskWriteIops, - ConfigDiskWriteFlow: &s.diskWriteFlow, - ConfigDiskWQueFactor: &s.diskWQueFactor, + ConfigDiskReadIocc: &s.diskReadIocc, + ConfigDiskReadIops: &s.diskReadIops, + ConfigDiskReadFlow: &s.diskReadFlow, + ConfigDiskWriteIocc: &s.diskWriteIocc, + ConfigDiskWriteIops: &s.diskWriteIops, + ConfigDiskWriteFlow: &s.diskWriteFlow, + ConfigDiskWQueFactor: &s.diskWQueFactor, + ConfigDiskAsyncReadIocc: &s.diskAsyncReadIocc, + ConfigDiskAsyncReadIops: &s.diskAsyncReadIops, + ConfigDiskAsyncReadFlow: &s.diskAsyncReadFlow, + ConfigDiskAsyncWriteIocc: &s.diskAsyncWriteIocc, + ConfigDiskAsyncWriteIops: &s.diskAsyncWriteIops, + ConfigDiskAsyncWriteFlow: &s.diskAsyncWriteFlow, + ConfigDiskDeleteIocc: &s.diskDeleteIocc, + ConfigDiskDeleteIops: &s.diskDeleteIops, + ConfigDiskDeleteFlow: &s.diskDeleteFlow, } { val, err, has := parser(key) if err != nil { @@ -363,13 +382,25 @@ func (s *DataNode) getDiskQos(w http.ResponseWriter, r *http.Request) { disks := make([]interface{}, 0) for _, diskItem := range s.space.GetDisks() { disk := &struct { - Path string `json:"path"` - Read util.LimiterStatus `json:"read"` - Write util.LimiterStatus `json:"write"` + Path string `json:"path"` + QosEnable bool `json:"qosEnable"` + AsyncQosEnable bool `json:"asyncQosEnable"` + Read util.LimiterStatus `json:"read"` + Write util.LimiterStatus `json:"write"` + AsyncRead util.LimiterStatus `json:"asyncRead"` + AsyncWrite util.LimiterStatus `json:"asyncWrite"` + Delete util.LimiterStatus `json:"delete"` + IopsStatus proto.IopsStatus `json:"IopsStatus"` }{ - Path: diskItem.Path, - Read: diskItem.limitRead.Status(false), - Write: diskItem.limitWrite.Status(false), + Path: diskItem.Path, + QosEnable: s.diskQosEnable || s.diskQosEnableFromMaster, + AsyncQosEnable: s.diskAsyncQosEnable, + Read: diskItem.limitRead.Status(false), + Write: diskItem.limitWrite.Status(false), + AsyncRead: diskItem.limitAsyncRead.Status(false), + AsyncWrite: diskItem.limitAsyncWrite.Status(false), + Delete: diskItem.limitDelete.Status(false), + IopsStatus: s.IopsStatus(), } disks = append(disks, disk) } diff --git a/datanode/storage/extent_store.go b/datanode/storage/extent_store.go index 29974fac4..e03a25f95 100644 --- a/datanode/storage/extent_store.go +++ b/datanode/storage/extent_store.go @@ -1211,7 +1211,6 @@ func (s *ExtentStore) GetAvailableTinyExtent() (extentID uint64, err error) { default: log.LogDebugf("dp %v GetAvailableTinyExtent not found", s.partitionID) return 0, NoAvailableExtentError - } } diff --git a/datanode/wrap_operator.go b/datanode/wrap_operator.go index 943f121bd..95ed4cd75 100644 --- a/datanode/wrap_operator.go +++ b/datanode/wrap_operator.go @@ -262,8 +262,7 @@ func (s *DataNode) handlePacketToCreateExtent(p *repl.Packet) { return } - partition.disk.allocCheckLimit(proto.IopsWriteType, 1) - partition.disk.limitWrite.Run(0, true, func() { + partition.disk.diskLimit(OpWrite, 0, func() { err = partition.ExtentStore().Create(p.ExtentID) }) } @@ -779,8 +778,7 @@ func (s *DataNode) handleMarkDeletePacket(p *repl.Packet, c net.Conn) { if err == nil { log.LogInfof("handleMarkDeletePacket Delete PartitionID(%v)_Extent(%v)_Offset(%v)_Size(%v)", p.PartitionID, p.ExtentID, ext.ExtentOffset, ext.Size) - partition.disk.allocCheckLimit(proto.IopsWriteType, 1) - partition.disk.limitWrite.Run(0, true, func() { + partition.disk.diskLimit(OpDelete, 0, func() { log.LogInfof("[handleBatchMarkDeletePacket] vol(%v) dp(%v) mark delete extent(%v)", partition.config.VolName, partition.partitionID, p.ExtentID) err = partition.ExtentStore().MarkDelete(p.ExtentID, int64(ext.ExtentOffset), int64(ext.Size)) if err != nil { @@ -791,16 +789,13 @@ func (s *DataNode) handleMarkDeletePacket(p *repl.Packet, c net.Conn) { } else { log.LogInfof("handleMarkDeletePacket Delete PartitionID(%v)_Extent(%v)", p.PartitionID, p.ExtentID) - partition.disk.allocCheckLimit(proto.IopsWriteType, 1) - if rs := partition.disk.limitWrite.Run(0, true, func() { - log.LogInfof("[handleBatchMarkDeletePacket] vol(%v) dp(%v) mark delete extent(%v)", partition.config.VolName, partition.partitionID, p.ExtentID) + partition.disk.diskLimit(OpDelete, 0, func() { + log.LogInfof("[handleMarkDeletePacket] vol(%v) dp(%v) mark delete extent(%v)", partition.config.VolName, partition.partitionID, p.ExtentID) err = partition.ExtentStore().MarkDelete(p.ExtentID, 0, 0) if err != nil { log.LogErrorf("action[handleMarkDeletePacket]: failed to mark delete extent(%v), %v", p.ExtentID, err) } - }); err == nil && rs != nil { - err = rs - } + }) } } @@ -847,8 +842,7 @@ func (s *DataNode) handleBatchMarkDeletePacket(p *repl.Packet, c net.Conn) { } log.LogInfof(fmt.Sprintf("[handleBatchMarkDeletePacket] recive DeleteExtent (%v) from (%v)", ext, c.RemoteAddr().String())) - partition.disk.allocCheckLimit(proto.IopsWriteType, 1) - writable := partition.disk.limitWrite.TryRun(0, func() { + partition.disk.diskLimit(OpDelete, 0, func() { if storage.IsTinyExtent(ext.ExtentId) || ext.IsSnapshotDeletion { log.LogInfof("[handleBatchMarkDeletePacket] vol(%v) dp(%v) mark delete extent(%v), tinyExtent or snapDeletion", partition.config.VolName, partition.partitionID, ext.ExtentId) @@ -876,12 +870,6 @@ func (s *DataNode) handleBatchMarkDeletePacket(p *repl.Packet, c net.Conn) { } }) - if !writable { - log.LogInfof("[handleBatchMarkDeletePacket] delete limitIo reach(%v), remote (%v) try again.", deleteLimiteRater.Limit(), c.RemoteAddr().String()) - err = storage.LimitedIoError - return - } - // NOTE: reutrn if meet error if err != nil { return @@ -955,10 +943,7 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) { partitionIOMetric = exporter.NewTPCnt(MetricPartitionIOName) } - partition.disk.allocCheckLimit(proto.FlowWriteType, uint32(p.Size)) - partition.disk.allocCheckLimit(proto.IopsWriteType, 1) - - if writable := partition.disk.limitWrite.TryRun(int(p.Size), func() { + if writable := partition.disk.tryDiskLimit(OpWrite, uint32(p.Size), func() { param := &storage.WriteParam{ ExtentID: p.ExtentID, Offset: p.ExtentOffset, @@ -976,6 +961,7 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) { err = storage.LimitedIoError return } + if !shallDegrade { s.metrics.MetricIOBytes.AddWithLabels(int64(p.Size), metricPartitionIOLabels) partitionIOMetric.SetWithLabels(err, metricPartitionIOLabels) @@ -989,10 +975,7 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) { partitionIOMetric = exporter.NewTPCnt(MetricPartitionIOName) } - partition.disk.allocCheckLimit(proto.FlowWriteType, uint32(p.Size)) - partition.disk.allocCheckLimit(proto.IopsWriteType, 1) - - if writable := partition.disk.limitWrite.TryRun(int(p.Size), func() { + if writable := partition.disk.tryDiskLimit(OpWrite, uint32(p.Size), func() { param := &storage.WriteParam{ ExtentID: p.ExtentID, Offset: p.ExtentOffset, @@ -1010,6 +993,7 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) { err = storage.LimitedIoError return } + if !shallDegrade { s.metrics.MetricIOBytes.AddWithLabels(int64(p.Size), metricPartitionIOLabels) partitionIOMetric.SetWithLabels(err, metricPartitionIOLabels) @@ -1029,10 +1013,7 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) { partitionIOMetric = exporter.NewTPCnt(MetricPartitionIOName) } - partition.disk.allocCheckLimit(proto.FlowWriteType, uint32(currSize)) - partition.disk.allocCheckLimit(proto.IopsWriteType, 1) - - if writable := partition.disk.limitWrite.TryRun(currSize, func() { + if writable := partition.disk.tryDiskLimit(OpWrite, uint32(currSize), func() { param := &storage.WriteParam{ ExtentID: p.ExtentID, Offset: p.ExtentOffset + int64(offset), @@ -1050,6 +1031,7 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) { err = storage.LimitedIoError return } + if !shallDegrade { s.metrics.MetricIOBytes.AddWithLabels(int64(p.Size), metricPartitionIOLabels) partitionIOMetric.SetWithLabels(err, metricPartitionIOLabels) @@ -2271,6 +2253,12 @@ func (s *DataNode) handlePacketToDeleteLostDisk(p *repl.Packet) { return } + if !disk.isLost { + err = errors.NewErrorf("disk(%v) is not lost", request.DiskPath) + log.LogErrorf("action[handlePacketToDeleteLostDisk] disk(%v) is not lost", request.DiskPath) + return + } + s.space.deleteDisk(disk) log.LogInfof("action[handlePacketToDeleteLostDisk] delete lost disk (%v) success", request.DiskPath) } @@ -2303,6 +2291,18 @@ func (s *DataNode) handlePacketToReloadDisk(p *repl.Packet) { } log.LogWarnf("action[handlePacketToReloadDisk] try reload disk %v req %v", request.DiskPath, task.RequestID) + disk, err := s.space.GetDisk(request.DiskPath) + if err != nil { + log.LogErrorf("action[handlePacketToReloadDisk] disk(%v) is not found err(%v).", request.DiskPath, err) + return + } + + if !disk.isLost { + err = errors.NewErrorf("disk(%v) is not lost", request.DiskPath) + log.LogErrorf("action[handlePacketToReloadDisk] disk(%v) is not lost", request.DiskPath) + return + } + err = s.space.reloadDisk(request.DiskPath) if err != nil { log.LogErrorf("action[handlePacketToReloadDisk] disk(%v) reload failed, err (%v)", request.DiskPath, err) diff --git a/datanode/wrap_operator_test.go b/datanode/wrap_operator_test.go index 1508642ed..ce18edb8f 100644 --- a/datanode/wrap_operator_test.go +++ b/datanode/wrap_operator_test.go @@ -16,6 +16,7 @@ package datanode import ( "encoding/json" + "net" "os" "path" "sync" @@ -53,8 +54,11 @@ func newDiskForOperatorTest(t *testing.T, dn *DataNode) (d *Disk) { d.limitFactor[proto.FlowWriteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxFLowLimit), proto.QosDefaultBurst) d.limitFactor[proto.IopsReadType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) d.limitFactor[proto.IopsWriteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) + d.limitFactor[proto.IopsDeleteType] = rate.NewLimiter(rate.Limit(proto.QosDefaultDiskMaxIoLimit), defaultIOLimitBurst) d.limitRead = util.NewIOLimiter(1*util.MB, 10) d.limitWrite = util.NewIOLimiter(1*util.MB, 10) + d.limitAsyncRead = util.NewIOLimiter(1*util.MB, 10) + d.limitDelete = util.NewIOLimiter(1*util.MB, 10) return } @@ -89,6 +93,8 @@ func newDataNodeForOperatorTest(t *testing.T) (dn *DataNode) { metrics: &DataNodeMetrics{ dataNode: dn, }, + diskQosEnable: true, + diskAsyncQosEnable: true, } return } @@ -125,6 +131,115 @@ func newPacketForTest(task *proto.AdminTask) *repl.Packet { } } +func TestMarkDeleteIopsLimit(t *testing.T) { + var ( + wg sync.WaitGroup + c net.Conn + ) + + dn := newDataNodeForOperatorTest(t) + dp := newDpForOperatorTest(t, dn) + + for i := 100; i < 900; i++ { + p := newPacketForOperatorTest(t, dp, uint64(i)) + p.Opcode = proto.OpCreateExtent + dn.handlePacketToCreateExtent(p) + require.EqualValues(t, proto.OpOk, p.ResultCode) + } + + dp.dataNode.diskAsyncQosEnable = true + dp.disk.limitFactor[proto.IopsDeleteType].SetLimit(rate.Limit(200)) + startTime := time.Now() + for i := 100; i < 500; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + p := newPacketForOperatorTest(t, dp, uint64(i)) + p.Opcode = proto.OpMarkDelete + p.ExtentType = 1 + dn.handleMarkDeletePacket(p, c) + require.EqualValues(t, proto.OpOk, p.ResultCode) + }(i) + } + wg.Wait() + costTime1 := time.Since(startTime) + t.Logf("cost time1(%v)", costTime1) + + time.Sleep(time.Second) + + dp.disk.limitFactor[proto.IopsDeleteType].SetLimit(rate.Limit(50)) + startTime = time.Now() + for i := 500; i < 900; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + p := newPacketForOperatorTest(t, dp, uint64(i)) + p.Opcode = proto.OpMarkDelete + p.ExtentType = 1 + dn.handleMarkDeletePacket(p, c) + require.EqualValues(t, proto.OpOk, p.ResultCode) + }(i) + } + wg.Wait() + costTime2 := time.Since(startTime) + t.Logf("cost time2(%v)", costTime2) + + require.Greater(t, costTime2, costTime1) +} + +func TestMarkDeleteIoccLimit(t *testing.T) { + var ( + wg sync.WaitGroup + c net.Conn + ) + + dn := newDataNodeForOperatorTest(t) + dp := newDpForOperatorTest(t, dn) + + for i := 100; i < 700; i++ { + p := newPacketForOperatorTest(t, dp, uint64(i)) + p.Opcode = proto.OpCreateExtent + dn.handlePacketToCreateExtent(p) + require.EqualValues(t, proto.OpOk, p.ResultCode) + } + + dp.disk.limitFactor[proto.IopsDeleteType].SetLimit(rate.Limit(100)) + startTime := time.Now() + for i := 100; i < 400; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + p := newPacketForOperatorTest(t, dp, uint64(i)) + p.Opcode = proto.OpMarkDelete + p.ExtentType = 1 + dn.handleMarkDeletePacket(p, c) + }(i) + } + wg.Wait() + + costTime1 := time.Since(startTime) + t.Logf("cost time1(%v)", costTime1) + + dp.disk.limitDelete.ResetIO(2, 0) + startTime = time.Now() + for i := 400; i < 700; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + p := newPacketForOperatorTest(t, dp, uint64(i)) + p.Opcode = proto.OpMarkDelete + p.ExtentType = 1 + dn.handleMarkDeletePacket(p, c) + require.EqualValues(t, proto.OpOk, p.ResultCode) + }(i) + } + wg.Wait() + costTime2 := time.Since(startTime) + t.Logf("cost time2(%v)", costTime2) + + require.Greater(t, costTime2, costTime1) +} + func TestDeleteLostDisk(t *testing.T) { dn := &DataNode{ space: &SpaceManager{ @@ -134,31 +249,38 @@ func TestDeleteLostDisk(t *testing.T) { }, } - testDiskPath := "/test/disk1" + testDiskPath1 := "/test/disk1" + testDiskPath2 := "/test/disk2" - lostDisk := NewLostDisk( - testDiskPath, + lostDisk1 := NewLostDisk( + testDiskPath1, 1*util.TB, 0, 3, dn.space, true, ) - dn.space.putDisk(lostDisk) + lostDisk2 := NewLostDisk( + testDiskPath2, + 1*util.TB, + 0, + 3, + dn.space, + true, + ) + dn.space.putDisk(lostDisk1) + dn.space.putDisk(lostDisk2) t.Run("normal delete disk", func(t *testing.T) { - req := &proto.DeleteLostDiskRequest{DiskPath: testDiskPath} + req := &proto.DeleteLostDiskRequest{DiskPath: testDiskPath1} task := &proto.AdminTask{ OpCode: proto.OpDeleteLostDisk, Request: req, } p := newPacketForTest(task) - dn.handlePacketToDeleteLostDisk(p) - require.Equal(t, proto.OpOk, p.ResultCode) - - _, err := dn.space.GetDisk(testDiskPath) + _, err := dn.space.GetDisk(testDiskPath1) require.Error(t, err) require.Contains(t, err.Error(), "not exist") }) @@ -170,12 +292,24 @@ func TestDeleteLostDisk(t *testing.T) { Request: invalidReq, } p := newPacketForTest(task) - dn.handlePacketToDeleteLostDisk(p) - require.Equal(t, proto.OpIntraGroupNetErr, p.ResultCode) require.Contains(t, string(p.Data), "not exist") }) + + t.Run("delete unlost disk", func(t *testing.T) { + lostDisk2.isLost = false + invalidReq := &proto.DeleteLostDiskRequest{DiskPath: testDiskPath2} + task := &proto.AdminTask{ + OpCode: proto.OpDeleteLostDisk, + Request: invalidReq, + } + p := newPacketForTest(task) + dn.handlePacketToDeleteLostDisk(p) + require.Equal(t, proto.OpIntraGroupNetErr, p.ResultCode) + t.Logf("%v", string(p.Data)) + require.Contains(t, string(p.Data), "not lost") + }) } func TestReloadDisk(t *testing.T) { @@ -184,10 +318,12 @@ func TestReloadDisk(t *testing.T) { require.NoError(t, err) dn := &DataNode{ - diskReadFlow: 1 * util.MB, - diskWriteFlow: 1 * util.MB, - diskReadIocc: 10, - diskWriteIocc: 10, + diskReadFlow: 1 * util.MB, + diskAsyncReadFlow: 1 * util.MB, + diskWriteFlow: 1 * util.MB, + diskReadIocc: 10, + diskAsyncReadIocc: 10, + diskWriteIocc: 10, } sm := &SpaceManager{ disks: make(map[string]*Disk), @@ -211,13 +347,12 @@ func TestReloadDisk(t *testing.T) { ) dn.space.putDisk(disk) - req := &proto.ReloadDiskRequest{DiskPath: testDiskPath} - task := &proto.AdminTask{ - OpCode: proto.OpReloadDisk, - Request: req, - } - t.Run("normal reload disk", func(t *testing.T) { + req := &proto.ReloadDiskRequest{DiskPath: testDiskPath} + task := &proto.AdminTask{ + OpCode: proto.OpReloadDisk, + Request: req, + } p := newPacketForTest(task) dn.handlePacketToReloadDisk(p) require.Equal(t, proto.OpOk, p.ResultCode) @@ -227,38 +362,28 @@ func TestReloadDisk(t *testing.T) { }, 3*time.Second, 100*time.Millisecond, "disk not loaded") }) - t.Run("reload conflict", func(t *testing.T) { - var wg sync.WaitGroup - results := make(chan uint8, 2) - for i := 0; i < 2; i++ { - wg.Add(1) - go func() { - defer wg.Done() - p := newPacketForTest(&proto.AdminTask{ - OpCode: proto.OpReloadDisk, - Request: &proto.ReloadDiskRequest{DiskPath: testDiskPath}, - }) - dn.handlePacketToReloadDisk(p) - results <- p.ResultCode - }() + t.Run("reload unexist disk", func(t *testing.T) { + req := &proto.ReloadDiskRequest{DiskPath: "/invalid/path"} + task := &proto.AdminTask{ + OpCode: proto.OpReloadDisk, + Request: req, } + p := newPacketForTest(task) + dn.handlePacketToReloadDisk(p) + require.Equal(t, proto.OpIntraGroupNetErr, p.ResultCode) + require.Contains(t, string(p.Data), "not exist") + }) - go func() { - wg.Wait() - close(results) - }() - - var successCount, errorCount int - for res := range results { - switch res { - case proto.OpOk: - successCount++ - case proto.OpIntraGroupNetErr: - errorCount++ - } + t.Run("reload unlost disk", func(t *testing.T) { + disk.isLost = false + req := &proto.ReloadDiskRequest{DiskPath: testDiskPath} + task := &proto.AdminTask{ + OpCode: proto.OpReloadDisk, + Request: req, } - - require.Equal(t, 1, successCount) - require.Equal(t, 1, errorCount) + p := newPacketForTest(task) + dn.handlePacketToReloadDisk(p) + require.Equal(t, proto.OpIntraGroupNetErr, p.ResultCode) + require.Contains(t, string(p.Data), "not lost") }) } diff --git a/master/api_service.go b/master/api_service.go index dfa6b92e1..3db865166 100644 --- a/master/api_service.go +++ b/master/api_service.go @@ -1457,7 +1457,6 @@ func parseRequestQos(r *http.Request, isMagnify bool, isEnableIops bool) (qosPar } } } - log.LogInfof("action[parseRequestQos] result %v", qosParam) return @@ -8661,13 +8660,15 @@ func (m *Server) deleteLostDisk(w http.ResponseWriter, r *http.Request) { } } if !found { - sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: fmt.Sprintf("disk %v on %v is not lost ", diskPath, addr)}) + err = errors.NewErrorf("disk %v not found", diskPath) + sendErrReply(w, r, newErrHTTPReply(err)) return } - partitions := dataNode.badPartitions(diskPath, m.cluster) + partitions := dataNode.badPartitions(diskPath, m.cluster, true) if len(partitions) != 0 { - sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: fmt.Sprintf("disk %v still has partitions not decommission", diskPath)}) + err = errors.NewErrorf("disk %v not found", diskPath) + sendErrReply(w, r, newErrHTTPReply(err)) return } @@ -8705,14 +8706,22 @@ func (m *Server) reloadDisk(w http.ResponseWriter, r *http.Request) { return } - for _, path := range dataNode.AllDisks { + for _, path := range dataNode.LostDisks { if path == diskPath { found = true break } } if !found { - sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: fmt.Sprintf("disk %v not found", diskPath)}) + err = errors.NewErrorf("disk %v not found", diskPath) + sendErrReply(w, r, newErrHTTPReply(err)) + return + } + + partitions := dataNode.badPartitions(diskPath, m.cluster, true) + if len(partitions) != 0 { + err = errors.NewErrorf("disk %v still has partitions not decommission", diskPath) + sendErrReply(w, r, newErrHTTPReply(err)) return } diff --git a/master/api_service_test.go b/master/api_service_test.go index 5620dc219..aa1541e32 100644 --- a/master/api_service_test.go +++ b/master/api_service_test.go @@ -1997,7 +1997,7 @@ func TestDeleteLostDisk(t *testing.T) { func TestReloadDisk(t *testing.T) { addr := mds5Addr - disk := "/cfs/disk" + disk := "/cfs/disk2" reqUrl := fmt.Sprintf("%v%v?addr=%v&disk=%v", hostAddr, proto.ReloadDisk, addr, disk) process(reqUrl, t) diff --git a/master/cluster.go b/master/cluster.go index e6437fe60..cec97153d 100644 --- a/master/cluster.go +++ b/master/cluster.go @@ -4755,7 +4755,7 @@ func (c *Cluster) TryDecommissionDataNode(dataNode *DataNode) { var partitions []*DataPartition disks := dataNode.getDisks(c) for _, disk := range disks { - partitionsFromDisk := dataNode.badPartitions(disk, c) + partitionsFromDisk := dataNode.badPartitions(disk, c, false) partitions = append(partitions, partitionsFromDisk...) } // may allocate new dp when dataNode cancel decommission before @@ -5110,7 +5110,7 @@ func (c *Cluster) handleDataNodeBadDisk(dataNode *DataNode) { // TODO:no dp left on bad disk, notify sre to remove this disk // decommission failed, but lack replica for disk err dp is already removed retry := c.RetryDecommissionDisk(dataNode.Addr, disk.DiskPath) - partitions := dataNode.badPartitions(disk.DiskPath, c) + partitions := dataNode.badPartitions(disk.DiskPath, c, false) totalDpCnt := len(partitions) if totalDpCnt == 0 && !retry { // msg := fmt.Sprintf("disk(%v_%v) can be removed", dataNode.Addr, disk.DiskPath) @@ -5200,7 +5200,7 @@ func (c *Cluster) TryDecommissionDisk(disk *DecommissionDisk) { disk.markDecommissionFailed() return } - badPartitions = node.badPartitions(disk.DiskPath, c) + badPartitions = node.badPartitions(disk.DiskPath, c, false) for _, dp := range badPartitions { badPartitionIds = append(badPartitionIds, dp.PartitionID) } diff --git a/master/data_node.go b/master/data_node.go index 2bff6284d..7894ee0c4 100644 --- a/master/data_node.go +++ b/master/data_node.go @@ -137,14 +137,14 @@ func (dataNode *DataNode) checkLiveness() { } } -func (dataNode *DataNode) badPartitions(diskPath string, c *Cluster) (partitions []*DataPartition) { +func (dataNode *DataNode) badPartitions(diskPath string, c *Cluster, ignoreDiscard bool) (partitions []*DataPartition) { partitions = make([]*DataPartition, 0) vols := c.copyVols() if len(vols) == 0 { return partitions } for _, vol := range vols { - dps := vol.dataPartitions.checkBadDiskDataPartitions(diskPath, dataNode.Addr) + dps := vol.dataPartitions.checkBadDiskDataPartitions(diskPath, dataNode.Addr, ignoreDiscard) partitions = append(partitions, dps...) } return diff --git a/master/data_partition_map.go b/master/data_partition_map.go index 784101b90..3c2b65570 100644 --- a/master/data_partition_map.go +++ b/master/data_partition_map.go @@ -422,7 +422,7 @@ func (dpMap *DataPartitionMap) setAllDataPartitionsToReadOnly() { log.LogDebugf("action[setAllDataPartitionsToReadOnly] ReadWrite->ReadOnly dp cnt: %v", changedCnt) } -func (dpMap *DataPartitionMap) checkBadDiskDataPartitions(diskPath, nodeAddr string) (partitions []*DataPartition) { +func (dpMap *DataPartitionMap) checkBadDiskDataPartitions(diskPath, nodeAddr string, ignoreDiscard bool) (partitions []*DataPartition) { dpMapCache := make([]*DataPartition, 0) dpMap.RLock() for _, dp := range dpMap.partitionMap { @@ -432,6 +432,9 @@ func (dpMap *DataPartitionMap) checkBadDiskDataPartitions(diskPath, nodeAddr str partitions = make([]*DataPartition, 0) for _, dp := range dpMapCache { + if !ignoreDiscard && dp.IsDiscard { + continue + } if dp.containsBadDisk(diskPath, nodeAddr) { partitions = append(partitions, dp) } diff --git a/master/gapi_cluster.go b/master/gapi_cluster.go index d759a92b9..fbe28ea0e 100644 --- a/master/gapi_cluster.go +++ b/master/gapi_cluster.go @@ -188,7 +188,7 @@ func (m *ClusterService) decommissionDisk(ctx context.Context, args struct { return nil, err } - badPartitions := node.badPartitions(args.DiskPath, m.cluster) + badPartitions := node.badPartitions(args.DiskPath, m.cluster, false) if len(badPartitions) == 0 { err = fmt.Errorf("node[%v] disk[%v] does not have any data partition", node.Addr, args.DiskPath) return nil, err diff --git a/proto/admin_proto.go b/proto/admin_proto.go index 6ba27c05c..0aa886ea3 100644 --- a/proto/admin_proto.go +++ b/proto/admin_proto.go @@ -782,6 +782,14 @@ type QosToDataNode struct { QosFlowWriteLimit uint64 } +type IopsStatus struct { + ReadIops int + WriteIops int + AsyncReadIops int + AsyncWriteIops int + DeleteIops int +} + // MultiVersionOpRequest defines the request of type MultiVersionOpRequest struct { VolumeID string @@ -1204,10 +1212,16 @@ const ( ) const ( - IopsReadType uint32 = 0x01 - IopsWriteType uint32 = 0x02 - FlowReadType uint32 = 0x03 - FlowWriteType uint32 = 0x04 + IopsReadType uint32 = 0x01 + IopsWriteType uint32 = 0x02 + FlowReadType uint32 = 0x03 + FlowWriteType uint32 = 0x04 + IopsAsyncReadType uint32 = 0x05 + IopsAsyncWriteType uint32 = 0x06 + FlowAsyncReadType uint32 = 0x07 + FlowAsyncWriteType uint32 = 0x08 + IopsDeleteType uint32 = 0x09 + FlowDeleteType uint32 = 0x0A ) const (