mirror of
https://github.com/cubefs/cubefs.git
synced 2026-08-02 02:00:56 +00:00
feat(data): limitIO strategy for separating repair io from business io. #1000095860
Signed-off-by: zhumingze <zhumingze@oppo.com>
This commit is contained in:
zhumingze
zhumingze1108
parent
f4bcdc21c6
commit
71dee686ef
@ -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") }
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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{}{}
|
||||
}
|
||||
|
||||
125
datanode/disk.go
125
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()
|
||||
|
||||
@ -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))
|
||||
|
||||
@ -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),
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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")
|
||||
})
|
||||
}
|
||||
|
||||
@ -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
|
||||
}
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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)
|
||||
}
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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 (
|
||||
|
||||
Loading…
Reference in New Issue
Block a user
