diff --git a/cli/cmd/fmt.go b/cli/cmd/fmt.go index b7898bcb3..bc6147fdb 100644 --- a/cli/cmd/fmt.go +++ b/cli/cmd/fmt.go @@ -440,7 +440,7 @@ func formatDataReplica(indentation string, replica *proto.DataReplica, rowTable } var sb = strings.Builder{} sb.WriteString(fmt.Sprintf("%v- Addr : %v\n", indentation, replica.Addr)) - sb.WriteString(fmt.Sprintf("%v Used : %v\n", indentation, formatSize(replica.Used))) + sb.WriteString(fmt.Sprintf("%v Allocated : %v\n", indentation, formatSize(replica.Used))) sb.WriteString(fmt.Sprintf("%v Total : %v\n", indentation, formatSize(replica.Total))) sb.WriteString(fmt.Sprintf("%v IsLeader : %v\n", indentation, replica.IsLeader)) sb.WriteString(fmt.Sprintf("%v FileCount : %v\n", indentation, replica.FileCount)) @@ -495,8 +495,8 @@ func formatDataNodeDetail(dn *proto.DataNodeInfo, rowTable bool) string { sb.WriteString(fmt.Sprintf(" ID : %v\n", dn.ID)) sb.WriteString(fmt.Sprintf(" Address : %v\n", dn.Addr)) sb.WriteString(fmt.Sprintf(" Carry : %v\n", dn.Carry)) - sb.WriteString(fmt.Sprintf(" Used ratio : %v\n", dn.UsageRatio)) - sb.WriteString(fmt.Sprintf(" Used : %v\n", formatSize(dn.Used))) + sb.WriteString(fmt.Sprintf(" Allocated ratio : %v\n", dn.UsageRatio)) + sb.WriteString(fmt.Sprintf(" Allocated : %v\n", formatSize(dn.Used))) sb.WriteString(fmt.Sprintf(" Available : %v\n", formatSize(dn.AvailableSpace))) sb.WriteString(fmt.Sprintf(" Total : %v\n", formatSize(dn.Total))) sb.WriteString(fmt.Sprintf(" Zone : %v\n", dn.ZoneName)) @@ -524,7 +524,7 @@ func formatMetaNodeDetail(mn *proto.MetaNodeInfo, rowTable bool) string { sb.WriteString(fmt.Sprintf(" Carry : %v\n", mn.Carry)) sb.WriteString(fmt.Sprintf(" Threshold : %v\n", mn.Threshold)) sb.WriteString(fmt.Sprintf(" MaxMemAvailWeight : %v\n", formatSize(mn.MaxMemAvailWeight))) - sb.WriteString(fmt.Sprintf(" Used : %v\n", formatSize(mn.Used))) + sb.WriteString(fmt.Sprintf(" Allocated : %v\n", formatSize(mn.Used))) sb.WriteString(fmt.Sprintf(" Total : %v\n", formatSize(mn.Total))) sb.WriteString(fmt.Sprintf(" Zone : %v\n", mn.ZoneName)) sb.WriteString(fmt.Sprintf(" IsActive : %v\n", formatNodeStatus(mn.IsActive))) diff --git a/client/fs/super.go b/client/fs/super.go index 6cda7f498..0ce348a8a 100644 --- a/client/fs/super.go +++ b/client/fs/super.go @@ -320,7 +320,6 @@ func (s *Super) Statfs(ctx context.Context, req *fuse.StatfsRequest, resp *fuse. func (s *Super) ClusterName() string { return s.cluster } - func (s *Super) GetRate(w http.ResponseWriter, r *http.Request) { w.Write([]byte(s.ec.GetRate())) } @@ -392,7 +391,7 @@ func (s *Super) SetSuspend(w http.ResponseWriter, r *http.Request) { sockaddr := r.FormValue("sock") if sockaddr == "" { - err = fmt.Errorf("Need parameter 'sock' for IPC") + err = fmt.Errorf("NeedAfterAlloc parameter 'sock' for IPC") replyFail(w, r, err.Error()) return } diff --git a/client/fuse.go b/client/fuse.go index 736d57c09..50912ca9d 100644 --- a/client/fuse.go +++ b/client/fuse.go @@ -339,7 +339,7 @@ func main() { syslog.SetOutput(outputFile) if *configRestoreFuse { - syslog.Println("Need restore fuse") + syslog.Println("NeedAfterAlloc restore fuse") opt.NeedRestoreFuse = true } diff --git a/datanode/data_partition_repair.go b/datanode/data_partition_repair.go index 3a4a1319f..b62373bb9 100644 --- a/datanode/data_partition_repair.go +++ b/datanode/data_partition_repair.go @@ -235,6 +235,9 @@ func (dp *DataPartition) DoRepair(repairTasks []*DataPartitionRepairTask) { if dp.ExtentStore().IsDeletedNormalExtent(extentInfo.FileID) { continue } + + dp.disk.allocCheckLimit(proto.IopsWriteType, 1) + store.Create(extentInfo.FileID) } for _, extentInfo := range repairTasks[0].ExtentsToBeRepaired { @@ -583,6 +586,10 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *storage.ExtentInfo currRecoverySize = binary.BigEndian.Uint64(reply.Arg[1:9]) reply.Size = uint32(currRecoverySize) } + + dp.disk.allocCheckLimit(proto.FlowWriteType, uint32(reply.Size)) + dp.disk.allocCheckLimit(proto.IopsWriteType, 1) + err = store.TinyExtentRecover(uint64(localExtentInfo.FileID), int64(currFixOffset), int64(currRecoverySize), reply.Data, reply.CRC, isEmptyResponse) if hasRecoverySize+currRecoverySize >= remoteAvaliSize { log.LogInfof("streamRepairTinyExtent(%v) recover fininsh,remoteAvaliSize(%v) "+ @@ -591,6 +598,9 @@ func (dp *DataPartition) streamRepairExtent(remoteExtentInfo *storage.ExtentInfo break } } else { + dp.disk.allocCheckLimit(proto.FlowWriteType, uint32(reply.Size)) + dp.disk.allocCheckLimit(proto.IopsWriteType, 1) + err = store.Write(uint64(localExtentInfo.FileID), int64(currFixOffset), int64(reply.Size), reply.Data, reply.CRC, storage.AppendWriteType, BufferWrite) } diff --git a/datanode/disk.go b/datanode/disk.go index c5ff354ff..2439f470f 100644 --- a/datanode/disk.go +++ b/datanode/disk.go @@ -16,6 +16,8 @@ package datanode import ( "fmt" + "golang.org/x/net/context" + "golang.org/x/time/rate" "path" "regexp" "strconv" @@ -64,6 +66,8 @@ type Disk struct { syncTinyDeleteRecordFromLeaderOnEveryDisk chan bool space *SpaceManager dataNode *DataNode + + limitFactor map[uint32]*rate.Limiter } const ( @@ -86,9 +90,40 @@ func NewDisk(path string, reservedSpace, diskRdonlySpace uint64, maxErrCnt int, d.computeUsage() d.updateSpaceInfo() d.startScheduleToUpdateSpaceInfo() + + d.limitFactor = make(map[uint32]*rate.Limiter, 0) + d.limitFactor[proto.FlowReadType] = rate.NewLimiter(rate.Inf, proto.QosDefaultDiskMaxFLowLimit) + d.limitFactor[proto.FlowWriteType] = rate.NewLimiter(rate.Inf, proto.QosDefaultDiskMaxFLowLimit) + d.limitFactor[proto.IopsReadType] = rate.NewLimiter(rate.Inf, proto.QosDefaultDiskMaxIoLimit) + d.limitFactor[proto.IopsWriteType] = rate.NewLimiter(rate.Inf, proto.QosDefaultDiskMaxIoLimit) + return } +func (d *Disk) updateQosLimiter() { + if d.dataNode.diskFlowReadLimit > 0 { + d.limitFactor[proto.FlowReadType].SetLimit(rate.Limit(d.dataNode.diskFlowReadLimit)) + } + if d.dataNode.diskFlowWriteLimit > 0 { + d.limitFactor[proto.FlowWriteType].SetLimit(rate.Limit(d.dataNode.diskFlowWriteLimit)) + } + if d.dataNode.diskIopsReadLimit > 0 { + d.limitFactor[proto.IopsReadType].SetLimit(rate.Limit(d.dataNode.diskIopsReadLimit)) + } + if d.dataNode.diskIopsWriteLimit > 0 { + d.limitFactor[proto.IopsWriteType].SetLimit(rate.Limit(d.dataNode.diskIopsWriteLimit)) + } +} + +func (d *Disk) allocCheckLimit(factorType uint32, used uint32) error { + if !(d.dataNode.diskQosEnableFromMaster && d.dataNode.diskQosEnable) { + return nil + } + ctx := context.Background() + d.limitFactor[factorType].WaitN(ctx, int(used)) + return nil +} + // 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 a8801cd56..a9841d896 100644 --- a/datanode/partition.go +++ b/datanode/partition.go @@ -763,6 +763,9 @@ func (dp *DataPartition) DoExtentStoreRepair(repairTask *DataPartitionRepairTask log.LogWarnf("AutoRepairStatus is False,so cannot Create extent(%v)", extentInfo.String()) continue } + + dp.disk.allocCheckLimit(proto.IopsWriteType, 1) + err := store.Create(uint64(extentInfo.FileID)) if err != nil { continue @@ -887,6 +890,7 @@ 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)) } diff --git a/datanode/partition_op_by_raft.go b/datanode/partition_op_by_raft.go index 63ce17ae4..679f6383d 100644 --- a/datanode/partition_op_by_raft.go +++ b/datanode/partition_op_by_raft.go @@ -238,6 +238,9 @@ 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) + err = dp.ExtentStore().Write(opItem.extentID, opItem.offset, opItem.size, opItem.data, opItem.crc, storage.RandomWriteType, opItem.opcode == proto.OpSyncRandomWrite) if err == nil { break diff --git a/datanode/server.go b/datanode/server.go index 5265823ab..49baf9795 100644 --- a/datanode/server.go +++ b/datanode/server.go @@ -95,6 +95,9 @@ const ( ConfigKeySmuxMaxConn = "smuxMaxConn" //int ConfigKeySmuxStreamPerConn = "smuxStreamPerConn" //int ConfigKeySmuxMaxBuffer = "smuxMaxBuffer" //int + + //rate limit control enable + ConfigDiskQosEnable = "diskQosEnable" //bool ) // DataNode defines the structure of a data node. @@ -132,6 +135,13 @@ type DataNode struct { metricsCnt uint64 control common.Control + + diskQosEnable bool + diskQosEnableFromMaster bool + diskIopsReadLimit uint64 + diskIopsWriteLimit uint64 + diskFlowReadLimit uint64 + diskFlowWriteLimit uint64 } func NewServer() *DataNode { @@ -262,6 +272,16 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) { return } +func (s *DataNode) initQosLimit(cfg *config.Config) { + s.space.dataNode.diskQosEnable = cfg.GetBoolWithDefault(ConfigDiskQosEnable, true) + log.LogWarnf("action[initQosLimit] set qos value [%v] ,other param use default value", s.space.dataNode.diskQosEnable) +} + +func (s *DataNode) updateQosLimit() { + for _, disk := range s.space.disks { + disk.updateQosLimiter() + } +} func (s *DataNode) startSpaceManager(cfg *config.Config) (err error) { s.startTime = time.Now().Unix() s.space = NewSpaceManager(s) @@ -273,6 +293,7 @@ func (s *DataNode) startSpaceManager(cfg *config.Config) (err error) { s.space.SetRaftStore(s.raftStore) s.space.SetNodeID(s.nodeID) s.space.SetClusterID(s.clusterID) + s.initQosLimit(cfg) diskRdonlySpace := uint64(cfg.GetInt64(CfgDiskRdonlySpace)) if diskRdonlySpace < DefaultDiskRetainMin { @@ -313,6 +334,7 @@ func (s *DataNode) startSpaceManager(cfg *config.Config) (err error) { s.space.LoadDisk(path, reservedSpace, diskRdonlySpace, DefaultDiskMaxErr) }(&wg, path, reservedSpace) } + wg.Wait() return nil } @@ -448,6 +470,7 @@ func (s *DataNode) registerHandler() { http.HandleFunc("/getSmuxPoolStat", s.getSmuxPoolStat()) http.HandleFunc("/setMetricsDegrade", s.setMetricsDegrade) http.HandleFunc("/getMetricsDegrade", s.getMetricsDegrade) + http.HandleFunc("/qosEnable", s.setQosEnable()) } func (s *DataNode) startTCPService() (err error) { diff --git a/datanode/server_handler.go b/datanode/server_handler.go index 508cf68fb..23d1b989b 100644 --- a/datanode/server_handler.go +++ b/datanode/server_handler.go @@ -328,6 +328,25 @@ func (s *DataNode) getNormalDeleted(w http.ResponseWriter, r *http.Request) { return } +func (s *DataNode) setQosEnable() func(http.ResponseWriter, *http.Request) { + return func(w http.ResponseWriter, r *http.Request) { + var ( + err error + enable bool + ) + if err = r.ParseForm(); err != nil { + s.buildFailureResp(w, http.StatusBadRequest, err.Error()) + return + } + if enable, err = strconv.ParseBool(r.FormValue("enable")); err != nil { + s.buildFailureResp(w, http.StatusBadRequest, err.Error()) + return + } + s.diskQosEnable = enable + s.buildSuccessResp(w, "success") + } +} + func (s *DataNode) getSmuxPoolStat() func(http.ResponseWriter, *http.Request) { return func(w http.ResponseWriter, r *http.Request) { if !s.enableSmuxConnPool { diff --git a/datanode/wrap_operator.go b/datanode/wrap_operator.go index 7699beaa5..4ddb523d4 100644 --- a/datanode/wrap_operator.go +++ b/datanode/wrap_operator.go @@ -194,6 +194,9 @@ func (s *DataNode) handlePacketToCreateExtent(p *repl.Packet) { err = storage.BrokenDiskError return } + + partition.disk.allocCheckLimit(proto.IopsWriteType, 1) + err = partition.ExtentStore().Create(p.ExtentID) return @@ -267,6 +270,39 @@ func (s *DataNode) handleHeartbeatPacket(p *repl.Packet) { marshaled, _ := json.Marshal(task.Request) _ = json.Unmarshal(marshaled, request) response.Status = proto.TaskSucceeds + if s.diskQosEnableFromMaster != request.EnableDiskQos { + log.LogWarnf("action[handleHeartbeatPacket] master command disk qos enable change to [%v], local conf enable [%v]", + request.EnableDiskQos, + s.diskQosEnable) + } + s.diskQosEnableFromMaster = request.EnableDiskQos + + var needUpdate bool + if request.QosFlowWriteLimit > 0 && request.QosFlowWriteLimit != s.diskFlowWriteLimit { + s.diskFlowWriteLimit = request.QosFlowWriteLimit + needUpdate = true + } + if request.QosFlowReadLimit > 0 && request.QosFlowReadLimit != s.diskFlowReadLimit { + s.diskFlowReadLimit = request.QosFlowReadLimit + needUpdate = true + } + if request.QosIopsWriteLimit > 0 && request.QosIopsWriteLimit != s.diskIopsWriteLimit { + s.diskIopsWriteLimit = request.QosIopsWriteLimit + needUpdate = true + } + if request.QosIopsReadLimit > 0 && request.QosIopsReadLimit != s.diskIopsReadLimit { + s.diskIopsReadLimit = request.QosIopsReadLimit + needUpdate = true + } + + if needUpdate { + log.LogWarnf("action[handleHeartbeatPacket] master change disk qos limit to [%v, %v ,%v, %v]", + s.diskFlowWriteLimit, + s.diskFlowReadLimit, + s.diskIopsWriteLimit, + s.diskIopsReadLimit) + s.updateQosLimit() + } } else { response.Status = proto.TaskFailed err = fmt.Errorf("illegal opcode") @@ -387,11 +423,13 @@ 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.ExtentStore().MarkDelete(p.ExtentID, int64(ext.ExtentOffset), int64(ext.Size)) } } else { log.LogInfof("handleMarkDeletePacket Delete PartitionID(%v)_Extent(%v)", p.PartitionID, p.ExtentID) + partition.disk.allocCheckLimit(proto.IopsWriteType, 1) partition.ExtentStore().MarkDelete(p.ExtentID, 0, 0) } @@ -419,6 +457,7 @@ func (s *DataNode) handleBatchMarkDeletePacket(p *repl.Packet, c net.Conn) { for _, ext := range exts { if deleteLimiteRater.Allow() { log.LogInfof(fmt.Sprintf("recive DeleteExtent (%v) from (%v)", ext, c.RemoteAddr().String())) + partition.disk.allocCheckLimit(proto.IopsWriteType, 1) store.MarkDelete(ext.ExtentId, int64(ext.ExtentOffset), int64(ext.Size)) } else { log.LogInfof("delete limiter reach(%v), remote (%v) try again.", deleteLimiteRater.Limit(), c.RemoteAddr().String()) @@ -462,6 +501,10 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) { if !shallDegrade { partitionIOMetric = exporter.NewTPCnt(MetricPartitionIOName) } + + partition.disk.allocCheckLimit(proto.FlowWriteType, uint32(p.Size)) + partition.disk.allocCheckLimit(proto.IopsWriteType, 1) + err = store.Write(p.ExtentID, p.ExtentOffset, int64(p.Size), p.Data, p.CRC, storage.AppendWriteType, p.IsSyncWrite()) if !shallDegrade { s.metrics.MetricIOBytes.AddWithLabels(int64(p.Size), metricPartitionIOLabels) @@ -475,6 +518,10 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) { if !shallDegrade { partitionIOMetric = exporter.NewTPCnt(MetricPartitionIOName) } + + partition.disk.allocCheckLimit(proto.FlowWriteType, uint32(p.Size)) + partition.disk.allocCheckLimit(proto.IopsWriteType, 1) + err = store.Write(p.ExtentID, p.ExtentOffset, int64(p.Size), p.Data, p.CRC, storage.AppendWriteType, p.IsSyncWrite()) if !shallDegrade { s.metrics.MetricIOBytes.AddWithLabels(int64(p.Size), metricPartitionIOLabels) @@ -494,6 +541,10 @@ func (s *DataNode) handleWritePacket(p *repl.Packet) { if !shallDegrade { partitionIOMetric = exporter.NewTPCnt(MetricPartitionIOName) } + + partition.disk.allocCheckLimit(proto.FlowWriteType, uint32(currSize)) + partition.disk.allocCheckLimit(proto.IopsWriteType, 1) + err = store.Write(p.ExtentID, p.ExtentOffset+int64(offset), int64(currSize), data, crc, storage.AppendWriteType, p.IsSyncWrite()) if !shallDegrade { s.metrics.MetricIOBytes.AddWithLabels(int64(p.Size), metricPartitionIOLabels) @@ -652,6 +703,10 @@ func (s *DataNode) extentRepairReadPacket(p *repl.Packet, connect net.Conn, isRe reply.ExtentOffset = offset p.Size = currReadSize p.ExtentOffset = offset + + partition.Disk().allocCheckLimit(proto.IopsReadType, 1) + partition.Disk().allocCheckLimit(proto.FlowReadType, currReadSize) + reply.CRC, err = store.Read(reply.ExtentID, offset, int64(currReadSize), reply.Data, isRepairRead) if !shallDegrade { s.metrics.MetricIOBytes.AddWithLabels(int64(p.Size), metricPartitionIOLabels) diff --git a/datanode/wrap_prepare.go b/datanode/wrap_prepare.go index 104786257..2c8554987 100644 --- a/datanode/wrap_prepare.go +++ b/datanode/wrap_prepare.go @@ -118,7 +118,7 @@ func (s *DataNode) addExtentInfo(p *repl.Packet) error { } p.ExtentID, err = store.NextExtentID() if err != nil { - return fmt.Errorf("addExtentInfo partition %v alloc NextExtentId error %v", p.PartitionID, err) + return fmt.Errorf("addExtentInfo partition %v allocCheckLimit NextExtentId error %v", p.PartitionID, err) } } else if p.IsLeaderPacket() && p.IsMarkDeleteExtentOperation() && p.IsTinyExtentType() { record := new(proto.TinyExtentDeleteRecord) diff --git a/depends/bazil.org/fuse/fs/serve.go b/depends/bazil.org/fuse/fs/serve.go index 63e4be158..416ba5d78 100644 --- a/depends/bazil.org/fuse/fs/serve.go +++ b/depends/bazil.org/fuse/fs/serve.go @@ -406,7 +406,7 @@ type Server struct { freeHandle []fuse.HandleID nodeGen uint64 - // Used to ensure worker goroutines finish before Serve returns + // Allocated to ensure worker goroutines finish before Serve returns wg sync.WaitGroup } diff --git a/depends/jacobsa/fuse/samples/statfs/statfs_darwin_test.go b/depends/jacobsa/fuse/samples/statfs/statfs_darwin_test.go index 9845fc419..6a4f896bc 100644 --- a/depends/jacobsa/fuse/samples/statfs/statfs_darwin_test.go +++ b/depends/jacobsa/fuse/samples/statfs/statfs_darwin_test.go @@ -26,7 +26,7 @@ import ( // Sample output: // -// Filesystem 1024-blocks Used Available Capacity iused ifree %iused Mounted on +// Filesystem 1024-blocks Allocated Available Capacity iused ifree %iused Mounted on // fake@bucket 32 16 16 50% 0 0 100% /Users/jacobsa/tmp/mp // var gDfOutputRegexp = regexp.MustCompile(`^\S+\s+(\d+)\s+(\d+)\s+(\d+)\s+\d+%\s+\d+\s+\d+\s+\d+%.*$`) diff --git a/depends/jacobsa/fuse/samples/statfs/statfs_linux_test.go b/depends/jacobsa/fuse/samples/statfs/statfs_linux_test.go index 86532dd92..499add90e 100644 --- a/depends/jacobsa/fuse/samples/statfs/statfs_linux_test.go +++ b/depends/jacobsa/fuse/samples/statfs/statfs_linux_test.go @@ -26,7 +26,7 @@ import ( // Sample output: // -// Filesystem 1K-blocks Used Available Use% Mounted on +// Filesystem 1K-blocks Allocated Available Use% Mounted on // some_fuse_file_system 512 64 384 15% /tmp/sample_test001288095 // var gDfOutputRegexp = regexp.MustCompile(`^\S+\s+(\d+)\s+(\d+)\s+(\d+)\s+\d+%.*$`) diff --git a/depends/spf13/cobra/cobra/cmd/root.go b/depends/spf13/cobra/cobra/cmd/root.go index 97f404bbb..20d62a9cb 100644 --- a/depends/spf13/cobra/cobra/cmd/root.go +++ b/depends/spf13/cobra/cobra/cmd/root.go @@ -22,7 +22,7 @@ import ( ) var ( - // Used for flags. + // Allocated for flags. cfgFile string userLicense string diff --git a/depends/spf13/cobra/command.go b/depends/spf13/cobra/command.go index ce9f5efe5..574efe914 100644 --- a/depends/spf13/cobra/command.go +++ b/depends/spf13/cobra/command.go @@ -366,7 +366,7 @@ func (c *Command) UsageFunc() (f func(*Command) error) { } // Usage puts out the usage for the command. -// Used when a user provides invalid input. +// Allocated when a user provides invalid input. // Can be defined by user by overriding UsageFunc. func (c *Command) Usage() error { return c.UsageFunc()(c) @@ -393,7 +393,7 @@ func (c *Command) HelpFunc() func(*Command, []string) { } // Help puts out the help for the command. -// Used when a user calls help [command]. +// Allocated when a user calls help [command]. // Can be defined by user by overriding HelpFunc. func (c *Command) Help() error { c.HelpFunc()(c, []string{}) diff --git a/depends/spf13/cobra/doc/man_docs.go b/depends/spf13/cobra/doc/man_docs.go index bd986eb99..0bc1b78d0 100644 --- a/depends/spf13/cobra/doc/man_docs.go +++ b/depends/spf13/cobra/doc/man_docs.go @@ -79,7 +79,7 @@ func GenManTreeFromOpts(cmd *cobra.Command, opts GenManTreeOptions) error { } // GenManTreeOptions is the options for generating the man pages. -// Used only in GenManTreeFromOpts. +// Allocated only in GenManTreeFromOpts. type GenManTreeOptions struct { Header *GenManHeader Path string diff --git a/docker/conf/client.json b/docker/conf/client.json index daa400297..15d71de32 100644 --- a/docker/conf/client.json +++ b/docker/conf/client.json @@ -4,7 +4,7 @@ "volName": "ltptest", "owner": "ltptest", "logDir": "/cfs/log", - "logLevel": "info", + "logLevel": "debug", "consulAddr": "http://192.168.0.101:8500", "exporterPort": 9500, "profPort": "17410", diff --git a/docker/conf/datanode.json b/docker/conf/datanode.json index 09dd6ff04..f00fafd2d 100644 --- a/docker/conf/datanode.json +++ b/docker/conf/datanode.json @@ -13,6 +13,12 @@ "disks": [ "/cfs/disk:10737418240" ], + + "diskIopsReadLimit" : "20000", + "diskIopsWriteLimit" : "5000", + "diskFlowReadLimit" : "1024000000", + "diskFlowWriteLimit": "524288000", + "masterAddr": [ "192.168.0.11:17010", "192.168.0.12:17010", diff --git a/docker/conf/master1.json b/docker/conf/master1.json index 82cee81ee..4c7768bb4 100644 --- a/docker/conf/master1.json +++ b/docker/conf/master1.json @@ -13,5 +13,7 @@ "logDir": "/cfs/log", "walDir": "/cfs/data/wal", "storeDir": "/cfs/data/store", - "metaNodeReservedMem": "67108864" + "metaNodeReservedMem": "67108864", + "fake_ebsAddr":"192.168.0.233:8500", + "fake_ebsServicePath":"access" } diff --git a/docker/conf/master2.json b/docker/conf/master2.json index 868315d95..933a61a4e 100644 --- a/docker/conf/master2.json +++ b/docker/conf/master2.json @@ -13,5 +13,7 @@ "logDir": "/cfs/log", "walDir": "/cfs/data/wal", "storeDir": "/cfs/data/store", - "metaNodeReservedMem": "67108864" + "metaNodeReservedMem": "67108864", + "fake_ebsAddr":"192.168.0.233:8500", + "fake_ebsServicePath":"access" } diff --git a/docker/conf/master3.json b/docker/conf/master3.json index 2d0ae8cf9..0b99b574e 100644 --- a/docker/conf/master3.json +++ b/docker/conf/master3.json @@ -13,5 +13,7 @@ "logDir": "/cfs/log", "walDir": "/cfs/data/wal", "storeDir": "/cfs/data/store", - "metaNodeReservedMem": "67108864" + "metaNodeReservedMem": "67108864", + "fake_ebsAddr":"192.168.0.233:8500", + "fake_ebsServicePath":"access" } diff --git a/libsdk/libsdk.go b/libsdk/libsdk.go index a1b2f9f40..d5e3f51f0 100644 --- a/libsdk/libsdk.go +++ b/libsdk/libsdk.go @@ -1109,18 +1109,20 @@ func (c *client) start() (err error) { c.bc = bcache.NewBcacheClient() } var ebsc *blobstore.BlobStoreClient - if ebsc, err = blobstore.NewEbsClient(access.Config{ - ConnMode: access.NoLimitConnMode, - Consul: api.Config{ - Address: c.ebsEndpoint, - }, - MaxSizePutOnce: MaxSizePutOnce, - Logger: &access.Logger{ - Filename: gopath.Join(c.logDir, "libcfs/ebs.log"), - }, - }); err != nil { - return - } + if c.ebsEndpoint != "" { + if ebsc, err = blobstore.NewEbsClient(access.Config{ + ConnMode: access.NoLimitConnMode, + Consul: api.Config{ + Address: c.ebsEndpoint, + }, + MaxSizePutOnce: MaxSizePutOnce, + Logger: &access.Logger{ + Filename: gopath.Join(c.logDir, "libcfs/ebs.log"), + }, + }); err != nil { + return + } + } var mw *meta.MetaWrapper if mw, err = meta.NewMetaWrapper(&meta.MetaConfig{ Volume: c.volName, diff --git a/master/api_args_parse.go b/master/api_args_parse.go index b25aaecf7..ec6a7e3ef 100644 --- a/master/api_args_parse.go +++ b/master/api_args_parse.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "hash/crc32" "io/ioutil" "net/http" "strconv" @@ -401,6 +402,19 @@ func parseRequestToSetVolCapacity(r *http.Request) (name, authKey string, capaci return } +type qosArgs struct { + qosEnable bool + diskQosEnable bool + iopsRVal uint64 + iopsWVal uint64 + flowRVal uint64 + flowWVal uint64 +} + +func (qos *qosArgs) isArgsWork() bool { + return (qos.iopsRVal | qos.iopsWVal | qos.flowRVal | qos.flowWVal) > 0 +} + type coldVolArgs struct { objBlockSize int cacheCap uint64 @@ -414,21 +428,23 @@ type coldVolArgs struct { } type createVolReq struct { - name string - owner string - size int - mpCount int - dpReplicaNum int - capacity int - followerRead bool - authenticate bool - crossZone bool - normalZonesFirst bool - domainId uint64 - zoneName string - description string - volType int - enablePosixAcl bool + name string + owner string + size int + mpCount int + dpReplicaNum int + capacity int + followerRead bool + authenticate bool + crossZone bool + normalZonesFirst bool + domainId uint64 + zoneName string + description string + volType int + enablePosixAcl bool + qosLimitArgs *qosArgs + clientReqPeriod, clientHitTriggerCnt uint32 // cold vol args coldArgs coldVolArgs } @@ -542,6 +558,9 @@ func parseRequestToCreateVol(r *http.Request, req *createVolReq) (err error) { return } + if req.qosLimitArgs, err = parseRequestQos(r, false); err != nil { + return err + } req.zoneName = extractStr(r, zoneNameKey) req.description = extractStr(r, descriptionKey) req.domainId, err = extractUint64WithDefault(r, domainIdKey, 0) @@ -925,6 +944,17 @@ func parseVolStatReq(r *http.Request) (name string, ver int, err error) { return } +func parseQosInfo(r *http.Request) (info *proto.ClientReportLimitInfo, err error) { + info = proto.NewClientReportLimitInfo() + var body []byte + if body, err = ioutil.ReadAll(r.Body); err != nil { + return + } + log.LogInfof("action[parseQosInfo] body len:[%v],crc:[%v]", len(body), crc32.ChecksumIEEE(body)) + err = json.Unmarshal(body, info) + return +} + func parseAndExtractName(r *http.Request) (name string, err error) { if err = r.ParseForm(); err != nil { return diff --git a/master/api_service.go b/master/api_service.go index d1b7d27d8..df5432c62 100644 --- a/master/api_service.go +++ b/master/api_service.go @@ -364,6 +364,310 @@ func (m *Server) createPreLoadDataPartition(w http.ResponseWriter, r *http.Reque cv.DataPartitions = dpResps sendOkReply(w, r, newSuccessHTTPReply(cv)) } +func (m *Server) getQosStatus(w http.ResponseWriter, r *http.Request) { + var ( + volName string + err error + vol *Vol + ) + if volName, err = extractName(r); err != nil { + sendErrReply(w, r, newErrHTTPReply(proto.ErrVolNotExists)) + return + } + if vol, err = m.cluster.getVol(volName); err != nil { + sendErrReply(w, r, newErrHTTPReply(proto.ErrVolNotExists)) + return + } + + sendOkReply(w, r, newSuccessHTTPReply(vol.getQosStatus())) +} + +func (m *Server) getClientQosInfo(w http.ResponseWriter, r *http.Request) { + var ( + volName string + err error + vol *Vol + host string + id uint64 + ) + if volName, err = extractName(r); err != nil { + sendErrReply(w, r, newErrHTTPReply(proto.ErrVolNotExists)) + return + } + if vol, err = m.cluster.getVol(volName); err != nil { + sendErrReply(w, r, newErrHTTPReply(proto.ErrVolNotExists)) + return + } + + if host = r.FormValue(addrKey); host != "" { + if !checkIp(host) { + sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: fmt.Errorf("addr not legal").Error()}) + return + } + } + + if value := r.FormValue(idKey); value != "" { + if id, err = strconv.ParseUint(value, 10, 64); err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + return + } + } + var rsp interface{} + if rsp, err = vol.getClientLimitInfo(id, util.GetIp(host)); err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + } else { + sendOkReply(w, r, newSuccessHTTPReply(rsp)) + } +} + +func (m *Server) QosUpdateClientParam(w http.ResponseWriter, r *http.Request) { + var ( + volName string + value string + period, triggerCnt uint64 + err error + vol *Vol + ) + if volName, err = extractName(r); err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + return + } + if vol, err = m.cluster.getVol(volName); err != nil { + sendErrReply(w, r, newErrHTTPReply(proto.ErrVolNotExists)) + return + } + if value = r.FormValue(ClientReqPeriod); value != "" { + if period, err = strconv.ParseUint(value, 10, 64); err != nil || period == 0 { + sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("wrong param of peroid"))) + } + } + if value = r.FormValue(ClientTriggerCnt); value != "" { + if triggerCnt, err = strconv.ParseUint(value, 10, 64); err != nil || triggerCnt == 0 { + sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("wrong param of triggerCnt"))) + } + } + if err = vol.updateClientParam(m.cluster, period, triggerCnt); err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + } + sendOkReply(w, r, newSuccessHTTPReply("success")) +} + +func parseRequestQos(r *http.Request, isMagnify bool) (qosParam *qosArgs, err error) { + qosParam = &qosArgs{} + + var value int + var flowFmt int + + if isMagnify { + flowFmt = 1 + } else { + flowFmt = util.MB + } + + if qosEnableStr := r.FormValue(QosEnableKey); qosEnableStr != "" { + qosParam.qosEnable, _ = strconv.ParseBool(qosEnableStr) + } + if iopsRLimitStr := r.FormValue(IopsRKey); iopsRLimitStr != "" { + log.LogInfof("actin[parseRequestQos] iopsRLimitStr %v", iopsRLimitStr) + if value, err = strconv.Atoi(iopsRLimitStr); err == nil { + qosParam.iopsRVal = uint64(value) + if !isMagnify && qosParam.iopsRVal < MinIoLimit { + err = fmt.Errorf("iops read %v need larger than 100", value) + return + } + if isMagnify && (qosParam.iopsRVal < MinMagnify || qosParam.iopsRVal > MaxMagnify) { + err = fmt.Errorf("iops read magnify %v must between %v and %v", value, MinMagnify, MaxMagnify) + log.LogErrorf("acttion[parseRequestQos] %v",err.Error()) + return + } + } + } + + if iopsWLimitStr := r.FormValue(IopsWKey); iopsWLimitStr != "" { + log.LogInfof("actin[parseRequestQos] iopsWLimitStr %v", iopsWLimitStr) + if value, err = strconv.Atoi(iopsWLimitStr); err == nil { + qosParam.iopsWVal = uint64(value) + if !isMagnify && qosParam.iopsWVal < MinIoLimit { + err = fmt.Errorf("iops %v write write io larger than 100", value) + return + } + if isMagnify && (qosParam.iopsWVal < MinMagnify || qosParam.iopsWVal > MaxMagnify) { + err = fmt.Errorf("iops write magnify %v must between %v and %v", value, MinMagnify, MaxMagnify) + log.LogErrorf("acttion[parseRequestQos] %v",err.Error()) + return + } + } + } + + if flowRLimitStr := r.FormValue(FlowRKey); flowRLimitStr != "" { + log.LogInfof("actin[parseRequestQos] flowRLimitStr %v", flowRLimitStr) + if value, err = strconv.Atoi(flowRLimitStr); err == nil { + qosParam.flowRVal = uint64(value * flowFmt) + if !isMagnify && qosParam.flowRVal < MinFlowLimit { + err = fmt.Errorf("flow read %v must larger than 100", value) + return + } + if isMagnify && (qosParam.flowRVal < MinMagnify || qosParam.flowRVal > MaxMagnify) { + err = fmt.Errorf("flow read magnify %v must between %v and %v", value, MinMagnify, MaxMagnify) + log.LogErrorf("acttion[parseRequestQos] %v",err.Error()) + return + } + } + } + if flowWLimitStr := r.FormValue(FlowWKey); flowWLimitStr != "" { + log.LogInfof("actin[parseRequestQos] flowWLimitStr %v", flowWLimitStr) + if value, err = strconv.Atoi(flowWLimitStr); err == nil { + qosParam.flowWVal = uint64(value * flowFmt) + if !isMagnify && qosParam.flowWVal < MinFlowLimit { + err = fmt.Errorf("flow write %v must larger than 100", value) + log.LogErrorf("acttion[parseRequestQos] %v",err.Error()) + return + } + if isMagnify && (qosParam.flowWVal < MinMagnify || qosParam.flowWVal > MaxMagnify) { + err = fmt.Errorf("flow write magnify %v must between %v and %v", value, MinMagnify, MaxMagnify) + log.LogErrorf("acttion[parseRequestQos] %v",err.Error()) + return + } + } + } + + log.LogInfof("action[parseRequestQos] result %v", qosParam) + + return +} + +func (m *Server) QosUpdateMagnify(w http.ResponseWriter, r *http.Request) { + var ( + volName string + err error + vol *Vol + magnifyArgs *qosArgs + ) + if volName, err = extractName(r); err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + return + } + + if vol, err = m.cluster.getVol(volName); err == nil { + if magnifyArgs, err = parseRequestQos(r, true); err == nil { + _ = vol.volQosUpdateMagnify(m.cluster, magnifyArgs) + sendOkReply(w, r, newSuccessHTTPReply("success")) + } + } + sendErrReply(w, r, newErrHTTPReply(err)) +} + +// flowRVal, flowWVal take MB as unit +func (m *Server) QosUpdateZoneLimit(w http.ResponseWriter, r *http.Request) { + var ( + value interface{} + ok bool + err error + qosParam *qosArgs + ) + var zoneName string + if zoneName = r.FormValue(zoneNameKey); zoneName == "" { + zoneName = DefaultZoneName + } + if qosParam, err = parseRequestQos(r, false); err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + return + } + + if value, ok = m.cluster.t.zoneMap.Load(zoneName); !ok { + sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("zonename [%v] not found", zoneName))) + return + } + zone := value.(*Zone) + zone.updateDataNodeQosLimit(m.cluster, qosParam) + + sendOkReply(w, r, newSuccessHTTPReply("success")) +} + +// flowRVal, flowWVal take MB as unit +func (m *Server) QosGetZoneLimit(w http.ResponseWriter, r *http.Request) { + var ( + value interface{} + ok bool + ) + var zoneName string + if zoneName = r.FormValue(zoneNameKey); zoneName == "" { + zoneName = DefaultZoneName + } + + if value, ok = m.cluster.t.zoneMap.Load(zoneName); !ok { + sendErrReply(w, r, newErrHTTPReply(fmt.Errorf("zonename [%v] not found", zoneName))) + return + } + zone := value.(*Zone) + + type qosZoneStatus struct { + Zone string + DiskLimitEnable bool + IopsRVal uint64 + IopsWVal uint64 + FlowRVal uint64 + FlowWVal uint64 + } + + zoneSt := &qosZoneStatus{ + Zone:zoneName, + DiskLimitEnable: m.cluster.diskQosEnable, + IopsRVal: zone.QosIopsRLimit, + IopsWVal: zone.QosIopsWLimit, + FlowRVal: zone.QosFlowRLimit, + FlowWVal: zone.QosFlowWLimit, + } + sendOkReply(w, r, newSuccessHTTPReply(zoneSt)) +} + +func (m *Server) QosUpdate(w http.ResponseWriter, r *http.Request) { + var ( + volName string + err error + vol *Vol + enable bool + value string + limitArgs *qosArgs + ) + if volName, err = extractName(r); err == nil { + if vol, err = m.cluster.getVol(volName); err != nil { + goto RET + } + if value = r.FormValue(QosEnableKey); value != "" { + if enable, err = strconv.ParseBool(value); err != nil { + goto RET + } + + if err = vol.volQosEnable(m.cluster, enable); err != nil { + goto RET + } + log.LogInfof("action[DiskQosUpdate] update qos eanble [%v]", enable) + } + if limitArgs, err = parseRequestQos(r, false); err == nil && limitArgs.isArgsWork() { + if err = vol.volQosUpdateLimit(m.cluster, limitArgs); err != nil { + goto RET + } + log.LogInfof("action[DiskQosUpdate] update qos limit [%v] [%v] [%v] [%v] [%v]", enable, + limitArgs.iopsRVal, limitArgs.iopsWVal, limitArgs.flowRVal, limitArgs.flowWVal) + } + } + + if value = r.FormValue(DiskEnableKey); value != "" { + if enable, err = strconv.ParseBool(value); err == nil { + log.LogInfof("action[DiskQosUpdate] enable be set [%v]", enable) + m.cluster.diskQosEnable = enable + err = m.cluster.syncPutCluster() + } + } +RET: + if err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + return + } + _ = sendOkReply(w, r, newSuccessHTTPReply("success")) + return +} func (m *Server) createDataPartition(w http.ResponseWriter, r *http.Request) { var ( @@ -975,12 +1279,54 @@ func (m *Server) createVol(w http.ResponseWriter, r *http.Request) { sendOkReply(w, r, newSuccessHTTPReply(msg)) } +func (m *Server) qosUpload(w http.ResponseWriter, r *http.Request) { + var ( + err error + name string + vol *Vol + limit *proto.LimitRsp2Client + ) + if name, err = parseAndExtractName(r); err != nil { + sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()}) + return + } + + if vol, err = m.cluster.getVol(name); err != nil { + sendErrReply(w, r, newErrHTTPReply(proto.ErrVolNotExists)) + return + } + + qosEnableStr := r.FormValue(QosEnableKey) + if qosEnableStr == "" { + sendErrReply(w, r, newErrHTTPReply(proto.ErrParamError)) + return + } + // qos upload may called by client init,thus use qosEnable param to identify it weather need to calc by master + log.LogInfof("action[qosUpload] qosEnableStr:[%v]", qosEnableStr) + if qosEnable, _ := strconv.ParseBool(qosEnableStr); qosEnable { + log.LogInfof("action[qosUpload] qosEnableStr:[%v] qosEnabled", qosEnableStr) + if clientInfo, err := parseQosInfo(r); err == nil { + log.LogInfof("action[qosUpload] cliInfoMgrMap [%v],clientInfo id[%v] clientInfo.Host %v, remote addr", clientInfo.ID, clientInfo.Host, r.RemoteAddr) + if clientInfo.ID == 0 { + if limit, err = vol.qosManager.init(m.cluster, clientInfo.Host); err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + } + } else if limit, err = vol.qosManager.HandleClientQosReq(clientInfo, clientInfo.ID); err != nil { + sendErrReply(w, r, newErrHTTPReply(err)) + } + } else { + log.LogInfof("action[qosUpload] qosEnableStr:[%v] err [%v]", err) + } + } + + sendOkReply(w, r, newSuccessHTTPReply(limit)) +} + func (m *Server) getVolSimpleInfo(w http.ResponseWriter, r *http.Request) { var ( err error name string vol *Vol - volView *proto.SimpleVolView ) if name, err = parseAndExtractName(r); err != nil { @@ -993,7 +1339,8 @@ func (m *Server) getVolSimpleInfo(w http.ResponseWriter, r *http.Request) { return } - volView = newSimpleView(vol) + volView := newSimpleView(vol) + sendOkReply(w, r, newSuccessHTTPReply(volView)) } diff --git a/master/cluster.go b/master/cluster.go index 35abff5a8..7f9a6e956 100644 --- a/master/cluster.go +++ b/master/cluster.go @@ -62,6 +62,7 @@ type Cluster struct { zoneIdxMux sync.Mutex // zoneList []string followerReadManager *followerReadManager + diskQosEnable bool } type followerReadManager struct { @@ -153,6 +154,7 @@ func (c *Cluster) scheduleTask() { c.scheduleToUpdateStatInfo() c.scheduleToManageDp() c.scheduleToCheckVolStatus() + c.scheduleToCheckVolQos() c.scheduleToCheckDiskRecoveryProgress() c.scheduleToCheckMetaPartitionRecoveryProgress() c.scheduleToLoadMetaPartitions() @@ -283,6 +285,21 @@ func (c *Cluster) scheduleToCheckFollowerReadCache() { }() } +func (c *Cluster) scheduleToCheckVolQos() { + go func() { + //check vols after switching leader two minutes + for { + if c.partition.IsRaftLeader() { + vols := c.copyVols() + for _, vol := range vols { + vol.checkQos() + } + } + time.Sleep(time.Second * time.Duration(c.cfg.IntervalToCheckQos)) + } + }() +} + func (c *Cluster) scheduleToCheckNodeSetGrpManagerStatus() { go func() { for { @@ -401,7 +418,7 @@ func (c *Cluster) checkDataNodeHeartbeat() { c.dataNodes.Range(func(addr, dataNode interface{}) bool { node := dataNode.(*DataNode) node.checkLiveness() - task := node.createHeartbeatTask(c.masterAddr()) + task := node.createHeartbeatTask(c.masterAddr(), c.diskQosEnable) tasks = append(tasks, task) return true }) @@ -2453,6 +2470,12 @@ func (c *Cluster) doCreateVol(req *createVolReq) (vol *Vol, err error) { CacheLowWater: req.coldArgs.cacheLowWater, CacheLRUInterval: req.coldArgs.cacheLRUInterval, CacheRule: req.coldArgs.cacheRule, + + VolQosEnable: req.qosLimitArgs.qosEnable, + IopsRLimit: req.qosLimitArgs.iopsRVal, + IopsWLimit: req.qosLimitArgs.iopsWVal, + FlowRlimit: req.qosLimitArgs.flowRVal, + FlowWlimit: req.qosLimitArgs.flowWVal, } log.LogInfof("[doCreateVol] volView, %v", vv) diff --git a/master/config.go b/master/config.go index 2bde439b1..bebd1521a 100644 --- a/master/config.go +++ b/master/config.go @@ -51,6 +51,7 @@ const ( defaultIntervalToCheck = 60 defaultIntervalToCheckHeartbeat = 6 defaultIntervalToCheckDataPartition = 5 + defaultIntervalToCheckQos = 3 defaultIntervalToCheckCrc = 20 * defaultIntervalToCheck // in terms of seconds noHeartBeatTimes = 3 // number of times that no heartbeat reported defaultNodeTimeOutSec = noHeartBeatTimes * defaultIntervalToCheckHeartbeat @@ -86,6 +87,7 @@ type clusterConfig struct { PeriodToLoadALLDataPartitions int64 metaNodeReservedMem uint64 IntervalToCheckDataPartition int // seconds + IntervalToCheckQos int // seconds numberOfDataPartitionsToFree int numberOfDataPartitionsToLoad int nodeSetCapacity int @@ -115,6 +117,7 @@ func newClusterConfig() (cfg *clusterConfig) { cfg.MissingDataPartitionInterval = defaultMissingDataPartitionInterval cfg.DataPartitionTimeOutSec = defaultDataPartitionTimeOutSec cfg.IntervalToCheckDataPartition = defaultIntervalToCheckDataPartition + cfg.IntervalToCheckQos = defaultIntervalToCheckQos cfg.IntervalToAlarmMissingDataPartition = defaultIntervalToAlarmMissingDataPartition cfg.numberOfDataPartitionsToLoad = defaultNumberOfDataPartitionsToLoad cfg.PeriodToLoadALLDataPartitions = defaultPeriodToLoadAllDataPartitions diff --git a/master/const.go b/master/const.go index b6a4f0585..494f25db9 100644 --- a/master/const.go +++ b/master/const.go @@ -78,6 +78,14 @@ const ( forceKey = "force" raftForceDelKey = "raftForceDel" enablePosixAclKey = "enablePosixAcl" + QosEnableKey = "qosEnable" + DiskEnableKey = "diskenable" + IopsWKey = "iopsWKey" + IopsRKey = "iopsRKey" + FlowWKey = "flowWKey" + FlowRKey = "flowRKey" + ClientReqPeriod = "reqPeriod" + ClientTriggerCnt = "triggerCnt" ) const ( @@ -96,6 +104,10 @@ const ( const ( LRUCacheSize = 3 << 30 WriteBufferSize = 4 * util.MB + MinFlowLimit = 1 * util.MB + MinIoLimit = 100 + MinMagnify = 10 + MaxMagnify = 100 ) const ( @@ -125,6 +137,13 @@ const ( defaultMigrateDpCnt = 50 defaultMigrateMpCnt = 15 defaultMaxReplicaCnt = 16 + defaultIopsRLimit uint64 = 1 << 16 + defaultIopsWLimit uint64 = 1 << 16 + defaultFlowWLimit uint64 = 1 << 35 + defaultFlowRLimit uint64 = 1 << 35 + defaultLimitTypeCnt = 4 + defaultClientTriggerHitCnt = 1 + defaultClientReqPeriodSeconds = 1 ) const ( @@ -180,6 +199,7 @@ const ( opSyncNodeSetGrp uint32 = 0x1F opSyncDataPartitionsView uint32 = 0x20 opSyncExclueDomain uint32 = 0x23 + opSyncUpdateZone uint32 = 0x24 ) const ( @@ -193,6 +213,7 @@ const ( clusterAcronym = "c" nodeSetAcronym = "s" nodeSetGrpAcronym = "g" + zoneAcronym = "zone" domainAcronym = "zoneDomain" maxDataPartitionIDKey = keySeparator + "max_dp_id" maxMetaPartitionIDKey = keySeparator + "max_mp_id" @@ -206,6 +227,7 @@ const ( nodeSetPrefix = keySeparator + nodeSetAcronym + keySeparator nodeSetGrpPrefix = keySeparator + nodeSetGrpAcronym + keySeparator DomainPrefix = keySeparator + domainAcronym + keySeparator + zonePrefix = keySeparator + zoneAcronym + keySeparator akAcronym = "ak" userAcronym = "user" volUserAcronym = "voluser" diff --git a/master/data_node.go b/master/data_node.go index 63fac0e0d..18a0c1cd5 100644 --- a/master/data_node.go +++ b/master/data_node.go @@ -50,6 +50,10 @@ type DataNode struct { ToBeOffline bool RdOnly bool MigrateLock sync.RWMutex + QosIopsRLimit uint64 + QosIopsWLimit uint64 + QosFlowRLimit uint64 + QosFlowWLimit uint64 } func newDataNode(addr, zoneName, clusterID string) (dataNode *DataNode) { @@ -189,11 +193,16 @@ func (dataNode *DataNode) clean() { dataNode.TaskManager.exitCh <- struct{}{} } -func (dataNode *DataNode) createHeartbeatTask(masterAddr string) (task *proto.AdminTask) { +func (dataNode *DataNode) createHeartbeatTask(masterAddr string, enableDiskQos bool) (task *proto.AdminTask) { request := &proto.HeartBeatRequest{ CurrTime: time.Now().Unix(), MasterAddr: masterAddr, } + request.EnableDiskQos = enableDiskQos + request.QosIopsReadLimit = dataNode.QosIopsRLimit + request.QosIopsWriteLimit = dataNode.QosIopsWLimit + request.QosFlowReadLimit = dataNode.QosFlowRLimit + request.QosFlowWriteLimit = dataNode.QosFlowWLimit task = proto.NewAdminTask(proto.OpDataNodeHeartbeat, dataNode.Addr, request) return } diff --git a/master/data_partition_check.go b/master/data_partition_check.go index 67b3bcb6d..807bb840a 100644 --- a/master/data_partition_check.go +++ b/master/data_partition_check.go @@ -106,7 +106,7 @@ func (partition *DataPartition) checkReplicaNotHaveStatus(liveReplicas []*DataRe func (partition *DataPartition) checkReplicaEqualStatus(liveReplicas []*DataReplica, status int8) (equal bool) { for _, replica := range liveReplicas { if replica.Status != status { - log.LogInfof("action[checkReplicaEqualStatus] partition %v replica %v status %v dst status %v", + log.LogDebugf("action[checkReplicaEqualStatus] partition %v replica %v status %v dst status %v", partition.PartitionID, replica.Addr, replica.Status, status) return } diff --git a/master/gapi_volume.go b/master/gapi_volume.go index f66c719ad..3f7f824b4 100644 --- a/master/gapi_volume.go +++ b/master/gapi_volume.go @@ -176,6 +176,7 @@ func (s *VolumeService) createVolume(ctx context.Context, args struct { Name, Owner, ZoneName, Description string Capacity, DataPartitionSize, MpCount, DpReplicaNum uint64 FollowerRead, Authenticate, CrossZone, DefaultPriority bool + iopsRLimit, iopsWLimit, flowRlimit, flowWlimit uint64 }) (*Vol, error) { uid, per, err := permissions(ctx, ADMIN|USER) if err != nil { @@ -205,6 +206,7 @@ func (s *VolumeService) createVolume(ctx context.Context, args struct { description: args.Description, } vol, err := s.cluster.createVol(req) + if err != nil { return nil, err } diff --git a/master/http_server.go b/master/http_server.go index c38dd7c0b..7a1598116 100644 --- a/master/http_server.go +++ b/master/http_server.go @@ -108,8 +108,8 @@ func (m *Server) registerAPIRoutes(router *mux.Router) { us := &UserService{user: m.user, cluster: m.cluster} m.registerHandler(router, proto.AdminUserAPI, us.Schema()) - vs := &VolumeService{user: m.user, cluster: m.cluster} - m.registerHandler(router, proto.AdminVolumeAPI, vs.Schema()) + //vs := &VolumeService{user: m.user, cluster: m.cluster} + //m.registerHandler(router, proto.AdminVolumeAPI, vs.Schema()) // cluster management APIs router.NewRoute().Name(proto.AdminGetIP). @@ -186,6 +186,30 @@ func (m *Server) registerAPIRoutes(router *mux.Router) { router.NewRoute().Methods(http.MethodGet). Path(proto.ClientMetaPartition). HandlerFunc(m.getMetaPartition) + router.NewRoute().Methods(http.MethodGet). + Path(proto.QosUpload). + HandlerFunc(m.qosUpload) + router.NewRoute().Methods(http.MethodGet). + Path(proto.QosGetStatus). + HandlerFunc(m.getQosStatus) + router.NewRoute().Methods(http.MethodGet). + Path(proto.QosGetClientsLimitInfo). + HandlerFunc(m.getClientQosInfo) + router.NewRoute().Methods(http.MethodGet). + Path(proto.QosUpdate). + HandlerFunc(m.QosUpdate) + router.NewRoute().Methods(http.MethodGet). + Path(proto.QosUpdateZoneLimit). + HandlerFunc(m.QosUpdateZoneLimit) + router.NewRoute().Methods(http.MethodGet). + Path(proto.QosGetZoneLimitInfo). + HandlerFunc(m.QosGetZoneLimit) + router.NewRoute().Methods(http.MethodGet). + Path(proto.QosUpdateMagnify). + HandlerFunc(m.QosUpdateMagnify) + router.NewRoute().Methods(http.MethodGet). + Path(proto.QosUpdateClientParam). + HandlerFunc(m.QosUpdateClientParam) router.NewRoute().Methods(http.MethodGet, http.MethodPost). Path(proto.AdminCreateMetaPartition). HandlerFunc(m.createMetaPartition) diff --git a/master/master_manager.go b/master/master_manager.go index 067ef3bba..64ba3397c 100644 --- a/master/master_manager.go +++ b/master/master_manager.go @@ -151,6 +151,10 @@ func (m *Server) loadMetadata() { panic(err) } + if err = m.cluster.loadZoneValue(); err != nil { + panic(err) + } + if err = m.cluster.loadVols(); err != nil { panic(err) } diff --git a/master/metadata_fsm_op.go b/master/metadata_fsm_op.go index 9af693aad..8c66e10ad 100644 --- a/master/metadata_fsm_op.go +++ b/master/metadata_fsm_op.go @@ -42,6 +42,7 @@ type clusterValue struct { DataNodeAutoRepairLimitRate uint64 MaxDpCntLimit uint64 FaultDomain bool + DiskQosEnable bool } func newClusterValue(c *Cluster) (cv *clusterValue) { @@ -56,6 +57,7 @@ func newClusterValue(c *Cluster) (cv *clusterValue) { DisableAutoAllocate: c.DisableAutoAllocate, MaxDpCntLimit: c.cfg.MaxDpCntLimit, FaultDomain: c.FaultDomain, + DiskQosEnable: c.diskQosEnable, } return cv } @@ -146,18 +148,19 @@ type volValue struct { Owner string FollowerRead bool Authenticate bool - CrossZone bool - DomainOn bool - ZoneName string - OSSAccessKey string - OSSSecretKey string - CreateTime int64 - Description string - DpSelectorName string - DpSelectorParm string - DefaultPriority bool - DomainId uint64 - VolType int + + CrossZone bool + DomainOn bool + ZoneName string + OSSAccessKey string + OSSSecretKey string + CreateTime int64 + Description string + DpSelectorName string + DpSelectorParm string + DefaultPriority bool + DomainId uint64 + VolType int EbsBlkSize int CacheCapacity uint64 @@ -170,6 +173,11 @@ type volValue struct { CacheRule string EnablePosixAcl bool + VolQosEnable bool + DiskQosEnable bool + IopsRLimit, IopsWLimit, FlowRlimit, FlowWlimit uint64 + IopsRMagnify, IopsWMagnify, FlowRMagnify, FlowWMagnify uint32 + ClientReqPeriod, ClientHitTriggerCnt uint32 } func (v *volValue) Bytes() (raw []byte, err error) { @@ -178,7 +186,6 @@ func (v *volValue) Bytes() (raw []byte, err error) { } func newVolValue(vol *Vol) (vv *volValue) { - vv = &volValue{ ID: vol.ID, Name: vol.Name, @@ -202,16 +209,27 @@ func newVolValue(vol *Vol) (vv *volValue) { DefaultPriority: vol.defaultPriority, EnablePosixAcl: vol.enablePosixAcl, - VolType: vol.VolType, - EbsBlkSize: vol.EbsBlkSize, - CacheCapacity: vol.CacheCapacity, - CacheAction: vol.CacheAction, - CacheThreshold: vol.CacheThreshold, - CacheTTL: vol.CacheTTL, - CacheHighWater: vol.CacheHighWater, - CacheLowWater: vol.CacheLowWater, - CacheLRUInterval: vol.CacheLRUInterval, - CacheRule: vol.CacheRule, + VolType: vol.VolType, + EbsBlkSize: vol.EbsBlkSize, + CacheCapacity: vol.CacheCapacity, + CacheAction: vol.CacheAction, + CacheThreshold: vol.CacheThreshold, + CacheTTL: vol.CacheTTL, + CacheHighWater: vol.CacheHighWater, + CacheLowWater: vol.CacheLowWater, + CacheLRUInterval: vol.CacheLRUInterval, + CacheRule: vol.CacheRule, + VolQosEnable: vol.qosManager.qosEnable, + IopsRLimit: vol.qosManager.getQosLimit(bsProto.IopsReadType), + IopsWLimit: vol.qosManager.getQosLimit(bsProto.IopsWriteType), + FlowRlimit: vol.qosManager.getQosLimit(bsProto.FlowReadType), + FlowWlimit: vol.qosManager.getQosLimit(bsProto.FlowWriteType), + IopsRMagnify: vol.qosManager.getQosMagnify(bsProto.IopsReadType), + IopsWMagnify: vol.qosManager.getQosMagnify(bsProto.IopsWriteType), + FlowRMagnify: vol.qosManager.getQosMagnify(bsProto.FlowReadType), + FlowWMagnify: vol.qosManager.getQosMagnify(bsProto.FlowWriteType), + ClientReqPeriod: vol.qosManager.ClientReqPeriod, + ClientHitTriggerCnt: vol.qosManager.ClientHitTriggerCnt, } return @@ -385,7 +403,7 @@ func (c *Cluster) syncUpdateNodeSet(nset *nodeSet) (err error) { } func (c *Cluster) putNodeSetInfo(opType uint32, nset *nodeSet) (err error) { - log.LogInfof("action[putNodeSetInfo], type:[%v], ID:[%v], name:[%v]", opType, nset.ID, nset.zoneName) + log.LogInfof("action[putNodeSetInfo], type:[%v], gridId:[%v], name:[%v]", opType, nset.ID, nset.zoneName) metadata := new(RaftCmd) metadata.Op = opType metadata.K = nodeSetPrefix + strconv.FormatUint(nset.ID, 10) @@ -461,6 +479,22 @@ func (c *Cluster) syncDeleteVol(vol *Vol) (err error) { return c.syncPutVolInfo(opSyncDeleteVol, vol) } +func (c *Cluster) sycnPutZoneInfo(zone *Zone) error { + var err error + metadata := new(RaftCmd) + metadata.Op = opSyncUpdateZone + metadata.K = zonePrefix + zone.name + vv := zone.getFsmValue() + if vv.Name == "" { + vv.Name = DefaultZoneName + } + log.LogInfof("action[sycnPutZoneInfo] zone name %v", vv.Name) + if metadata.V, err = json.Marshal(vv); err != nil { + return errors.New(err.Error()) + } + return c.submit(metadata) +} + func (c *Cluster) syncPutVolInfo(opType uint32, vol *Vol) (err error) { metadata := new(RaftCmd) metadata.Op = opType @@ -607,6 +641,37 @@ func (c *Cluster) updateMaxDpCntLimit(val uint64) { maxDpCntOneNode = uint32(val) } +func (c *Cluster) loadZoneValue() (err error) { + var ok bool + result, err := c.fsm.store.SeekForPrefix([]byte(zonePrefix)) + if err != nil { + err = fmt.Errorf("action[loadZoneValue],err:%v", err.Error()) + return err + } + for _, value := range result { + cv := &zoneValue{} + if err = json.Unmarshal(value, cv); err != nil { + log.LogErrorf("action[loadZoneValue], unmarshal err:%v", err.Error()) + continue + } + var zoneInfo interface{} + if zoneInfo, ok = c.t.zoneMap.Load(cv.Name); !ok { + log.LogErrorf("action[loadZoneValue], zonename [%v] not found", cv.Name) + continue + } + zone := zoneInfo.(*Zone) + zone.QosFlowRLimit = cv.QosFlowRLimit + zone.QosIopsWLimit = cv.QosIopsWLimit + zone.QosFlowWLimit = cv.QosFlowWLimit + zone.QosIopsRLimit = cv.QosIopsRLimit + log.LogInfof("action[loadZoneValue] load zonename[%v] with limit [%v,%v,%v,%v]", + zone.name, cv.QosFlowRLimit, cv.QosIopsWLimit, cv.QosFlowWLimit, cv.QosIopsRLimit) + zone.loadDataNodeQosLimit() + } + + return +} + func (c *Cluster) loadClusterValue() (err error) { result, err := c.fsm.store.SeekForPrefix([]byte(clusterPrefix)) if err != nil { @@ -622,6 +687,7 @@ func (c *Cluster) loadClusterValue() (err error) { c.cfg.MetaNodeThreshold = cv.Threshold c.cfg.ClusterLoadFactor = cv.LoadFactor c.DisableAutoAllocate = cv.DisableAutoAllocate + c.diskQosEnable = cv.DiskQosEnable c.updateMetaNodeDeleteBatchCount(cv.MetaNodeDeleteBatchCount) c.updateMetaNodeDeleteWorkerSleepMs(cv.MetaNodeDeleteWorkerSleepMs) c.updateDataNodeDeleteLimitRate(cv.DataNodeDeleteLimitRate) @@ -916,7 +982,7 @@ func (c *Cluster) loadMetaPartitions() (err error) { continue } if vol.ID != mpv.VolID { - Warn(c.Name, fmt.Sprintf("action[loadMetaPartitions] has duplicate vol[%v],vol.ID[%v],mpv.VolID[%v]", mpv.VolName, vol.ID, mpv.VolID)) + Warn(c.Name, fmt.Sprintf("action[loadMetaPartitions] has duplicate vol[%v],vol.gridId[%v],mpv.VolID[%v]", mpv.VolName, vol.ID, mpv.VolID)) continue } for i := 0; i < len(mpv.Peers); i++ { @@ -964,7 +1030,7 @@ func (c *Cluster) loadDataPartitions() (err error) { continue } if vol.ID != dpv.VolID { - Warn(c.Name, fmt.Sprintf("action[loadDataPartitions] has duplicate vol[%v],vol.ID[%v],mpv.VolID[%v]", dpv.VolName, vol.ID, dpv.VolID)) + Warn(c.Name, fmt.Sprintf("action[loadDataPartitions] has duplicate vol[%v],vol.gridId[%v],mpv.VolID[%v]", dpv.VolName, vol.ID, dpv.VolID)) continue } for i := 0; i < len(dpv.Peers); i++ { diff --git a/master/topology.go b/master/topology.go index d89d9cf92..2a10f1fc4 100644 --- a/master/topology.go +++ b/master/topology.go @@ -1204,8 +1204,19 @@ type Zone struct { metaNodes *sync.Map nodeSetMap map[uint64]*nodeSet nsLock sync.RWMutex + QosIopsRLimit uint64 + QosIopsWLimit uint64 + QosFlowRLimit uint64 + QosFlowWLimit uint64 sync.RWMutex } +type zoneValue struct { + Name string + QosIopsRLimit uint64 + QosIopsWLimit uint64 + QosFlowRLimit uint64 + QosFlowWLimit uint64 +} func newZone(name string) (zone *Zone) { zone = &Zone{name: name} @@ -1229,6 +1240,16 @@ func printZonesName(zones []*Zone) string { return str } +func (zone *Zone) getFsmValue() *zoneValue { + return &zoneValue{ + Name: zone.name, + QosIopsRLimit: zone.QosIopsRLimit, + QosIopsWLimit: zone.QosIopsWLimit, + QosFlowRLimit: zone.QosFlowRLimit, + QosFlowWLimit: zone.QosFlowWLimit, + } +} + func (zone *Zone) setStatus(status int) { zone.status = status } @@ -1640,6 +1661,63 @@ func (zone *Zone) getAvailNodeHosts(nodeType uint32, excludeNodeSets []uint64, e return ns.getAvailMetaNodeHosts(excludeHosts, replicaNum) } +func (zone *Zone) updateDataNodeQosLimit(cluster *Cluster, qosParam *qosArgs) error { + var err error + if qosParam.flowRVal > 0 { + zone.QosFlowRLimit = qosParam.flowRVal + } + if qosParam.flowWVal > 0 { + zone.QosFlowWLimit = qosParam.flowWVal + } + if qosParam.iopsRVal > 0 { + zone.QosIopsRLimit = qosParam.iopsRVal + } + if qosParam.iopsWVal > 0 { + zone.QosIopsWLimit = qosParam.iopsWVal + } + + if err = cluster.sycnPutZoneInfo(zone); err != nil { + return err + } + zone.dataNodes.Range(func(key, value interface{}) bool { + dataNode := value.(*DataNode) + if qosParam.flowRVal > 0 { + dataNode.QosFlowRLimit = qosParam.flowRVal + } + if qosParam.flowWVal > 0 { + dataNode.QosFlowWLimit = qosParam.flowWVal + } + if qosParam.iopsRVal > 0 { + dataNode.QosIopsRLimit = qosParam.iopsRVal + } + if qosParam.iopsWVal > 0 { + dataNode.QosIopsWLimit = qosParam.iopsWVal + } + return true + }) + return nil +} + +func (zone *Zone) loadDataNodeQosLimit() { + + zone.dataNodes.Range(func(key, value interface{}) bool { + dataNode := value.(*DataNode) + if zone.QosFlowRLimit > 0 { + dataNode.QosFlowRLimit = zone.QosFlowRLimit + } + if zone.QosFlowWLimit > 0 { + dataNode.QosFlowWLimit = zone.QosFlowWLimit + } + if zone.QosIopsRLimit > 0 { + dataNode.QosIopsRLimit = zone.QosIopsRLimit + } + if zone.QosIopsWLimit > 0 { + dataNode.QosIopsWLimit = zone.QosIopsWLimit + } + return true + }) +} + func (zone *Zone) dataNodeCount() (len int) { zone.dataNodes.Range(func(key, value interface{}) bool { diff --git a/master/vol.go b/master/vol.go index 0f87edfb6..e80c09569 100644 --- a/master/vol.go +++ b/master/vol.go @@ -86,6 +86,7 @@ type Vol struct { dpSelectorName string dpSelectorParm string domainId uint64 + qosManager *QosCtrlManager volLock sync.RWMutex } @@ -127,6 +128,24 @@ func newVol(vv volValue) (vol *Vol) { vol.CacheLowWater = vv.CacheLowWater vol.CacheLRUInterval = vv.CacheLRUInterval vol.CacheRule = vv.CacheRule + + limitQosVal := &qosArgs{ + qosEnable: vv.VolQosEnable, + diskQosEnable: vv.DiskQosEnable, + iopsRVal: vv.IopsRLimit, + iopsWVal: vv.IopsWLimit, + flowRVal: vv.FlowRlimit, + flowWVal: vv.FlowWlimit, + } + vol.initQosManager(limitQosVal) + + magnifyQosVal := &qosArgs{ + iopsRVal: uint64(vv.IopsRMagnify), + iopsWVal: uint64(vv.IopsWMagnify), + flowRVal: uint64(vv.FlowWMagnify), + flowWVal: uint64(vv.FlowWMagnify), + } + vol.qosManager.volUpdateMagnify(magnifyQosVal) return } @@ -157,6 +176,45 @@ func (vol *Vol) getPreloadCapacity() uint64 { return uint64(float32(total) / overSoldFactor) } +func (vol *Vol) initQosManager(limitArgs *qosArgs) { + vol.qosManager = &QosCtrlManager{ + cliInfoMgrMap: make(map[uint64]*ClientInfoMgr, 0), + serverFactorLimitMap: make(map[uint32]*ServerFactorLimit, 0), + qosEnable: limitArgs.qosEnable, + vol: vol, + ClientHitTriggerCnt: defaultClientTriggerHitCnt, + ClientReqPeriod: defaultClientReqPeriodSeconds, + } + + if limitArgs.iopsRVal == 0 { + limitArgs.iopsRVal = defaultIopsRLimit + } + if limitArgs.iopsWVal == 0 { + limitArgs.iopsWVal = defaultIopsWLimit + } + if limitArgs.flowRVal == 0 { + limitArgs.flowRVal = defaultFlowRLimit + } + if limitArgs.flowWVal == 0 { + limitArgs.flowWVal = defaultFlowWLimit + } + arrLimit := [defaultLimitTypeCnt]uint64{limitArgs.iopsRVal, limitArgs.iopsWVal, limitArgs.flowRVal, limitArgs.flowWVal} + arrType := [defaultLimitTypeCnt]uint32{proto.IopsReadType, proto.IopsWriteType, proto.FlowReadType, proto.FlowWriteType} + + for i := 0; i < defaultLimitTypeCnt; i++ { + vol.qosManager.serverFactorLimitMap[arrType[i]] = &ServerFactorLimit{ + Name: proto.QosTypeString(arrType[i]), + Type: arrType[i], + Total: arrLimit[i], + Buffer: arrLimit[i], + requestCh: make(chan interface{}, 10240), + qosManager: vol.qosManager, + } + go vol.qosManager.serverFactorLimitMap[arrType[i]].dispatch() + } + +} + func (vol *Vol) refreshOSSSecure() (key, secret string) { vol.OSSAccessKey = util.RandomString(16, util.Numeric|util.LowerLetter|util.UpperLetter) vol.OSSSecretKey = util.RandomString(32, util.Numeric|util.LowerLetter|util.UpperLetter) @@ -604,6 +662,583 @@ func (vol *Vol) ebsUsedSpace() uint64 { return size } +type ServerFactorLimit struct { + Name string + Type uint32 + Total uint64 + Buffer uint64 // flowbuffer add with preallocate buffer equal with flowtotal + CliUsed uint64 + CliNeed uint64 + Allocated uint64 + NeedAfterAlloc uint64 + Magnify uint32 // for client allocation need magnify + LimitRate float32 + requestCh chan interface{} + done chan interface{} + qosManager *QosCtrlManager +} + +type ClientInfoMgr struct { + Cli *proto.ClientReportLimitInfo + Assign *proto.LimitRsp2Client + Time time.Time + ID uint64 + host string +} + +type qosRequestArgs struct { + clientID uint64 + factorType uint32 + clientReq *proto.ClientLimitInfo + lastClientInfo *proto.ClientLimitInfo + assignInfo *proto.ClientLimitInfo + rsp2Client *proto.ClientLimitInfo + wg *sync.WaitGroup +} + +type QosCtrlManager struct { + cliInfoMgrMap map[uint64]*ClientInfoMgr // cientid->client_reportinfo&&assign_limitinfo + serverFactorLimitMap map[uint32]*ServerFactorLimit // vol qos data for iops w/r and flow w/r + defaultClientCnt uint32 + qosEnable bool + ClientReqPeriod uint32 + ClientHitTriggerCnt uint32 + vol *Vol + sync.RWMutex +} + +func (qosManager *QosCtrlManager) volUpdateMagnify(magnifyArgs *qosArgs) { + defer qosManager.Unlock() + qosManager.Lock() + + log.LogWarnf("action[volUpdateMagnify] vol %v try set magnify iopsRVal[%v],iopsWVal[%v],flowRVal[%v],flowWVal[%v]", + qosManager.vol.Name, magnifyArgs.iopsRVal, magnifyArgs.iopsWVal, magnifyArgs.flowRVal, magnifyArgs.flowWVal) + + arrMagnify := [4]uint64{magnifyArgs.iopsRVal, magnifyArgs.iopsWVal, magnifyArgs.flowRVal, magnifyArgs.flowWVal} + for i := proto.IopsReadType; i <= proto.FlowWriteType; i++ { + magnify := qosManager.serverFactorLimitMap[i].Magnify + if uint64(magnify) != arrMagnify[i-1] && arrMagnify[i-1] > 0 { + qosManager.serverFactorLimitMap[i].Magnify = uint32(arrMagnify[i-1]) + log.LogWarnf("action[volUpdateMagnify] vol %v after update type [%v] magnify [%v] to [%v]", + qosManager.vol.Name, proto.QosTypeString(i), magnify, arrMagnify[i-1]) + } + } +} + +func (qosManager *QosCtrlManager) volUpdateLimit(limitArgs *qosArgs) { + defer qosManager.Unlock() + qosManager.Lock() + + log.LogWarnf("action[volUpdateLimit] vol %v try set limit iopsrlimit[%v],iopswlimit[%v],flowrlimit[%v],flowwlimit[%v]", + qosManager.vol.Name, limitArgs.iopsRVal, limitArgs.iopsWVal, limitArgs.flowRVal, limitArgs.flowWVal) + + if limitArgs.iopsWVal != 0 { + qosManager.serverFactorLimitMap[proto.IopsWriteType].Total = limitArgs.iopsWVal + } + if limitArgs.iopsRVal != 0 { + qosManager.serverFactorLimitMap[proto.IopsReadType].Total = limitArgs.iopsRVal + } + if limitArgs.flowWVal != 0 { + qosManager.serverFactorLimitMap[proto.FlowWriteType].Total = limitArgs.flowWVal + } + if limitArgs.flowRVal != 0 { + qosManager.serverFactorLimitMap[proto.FlowReadType].Total = limitArgs.flowRVal + } + + for i := proto.IopsReadType; i <= proto.FlowWriteType; i++ { + limitf := qosManager.serverFactorLimitMap[i] + log.LogWarnf("action[volUpdateLimit] vol [%v] after set type [%v] [%v,%v,%v,%v]", + qosManager.vol.Name, proto.QosTypeString(i), limitf.Allocated, limitf.NeedAfterAlloc, limitf.Total, limitf.Buffer) + } +} + +func (qosManager *QosCtrlManager) getQosMagnify(factorTYpe uint32) uint32 { + return qosManager.serverFactorLimitMap[factorTYpe].Magnify +} + +func (qosManager *QosCtrlManager) getQosLimit(factorTYpe uint32) uint64 { + return qosManager.serverFactorLimitMap[factorTYpe].Total +} + +func (qosManager *QosCtrlManager) initClientQosInfo(clientID uint64, host string) (limitRsp2Client *proto.LimitRsp2Client, err error) { + + log.LogInfof("action[initClientQosInfo] vol %v clientID %v host %v", qosManager.vol.Name, clientID, host) + clientInitInfo := proto.NewClientReportLimitInfo() + cliCnt := qosManager.defaultClientCnt + if cliCnt <= proto.QosDefaultClientCnt { + cliCnt = proto.QosDefaultClientCnt + } + if len(qosManager.serverFactorLimitMap) > int(cliCnt) { + cliCnt = uint32(len(qosManager.serverFactorLimitMap)) + } + + limitRsp2Client = proto.NewLimitRsp2Client() + limitRsp2Client.ID = clientID + limitRsp2Client.Enable = qosManager.qosEnable + + factorType := proto.IopsReadType + + defer qosManager.Unlock() + qosManager.Lock() + + for factorType <= proto.FlowWriteType { + var initLimit uint64 + serverLimit := qosManager.serverFactorLimitMap[factorType] + initLimit = serverLimit.Total / uint64(cliCnt) + + if serverLimit.Buffer > initLimit { + serverLimit.Buffer -= initLimit + serverLimit.Allocated += initLimit + } else { + serverLimit.Buffer = 0 + } + + clientInitInfo.FactorMap[factorType] = &proto.ClientLimitInfo{ + UsedLimit: initLimit, + UsedBuffer: 0, + Used: 0, + Need: 0, + } + cliInfo := clientInitInfo.FactorMap[factorType] + if factorType == proto.FlowWriteType || factorType == proto.FlowReadType { + if cliInfo.UsedLimit > 1*util.GB/8 { + cliInfo.UsedLimit = 1 * util.GB / 8 + } + } else { + if cliInfo.UsedLimit > 5000 { + cliInfo.UsedLimit = 5000 + } + } + + limitRsp2Client.FactorMap[factorType] = cliInfo + limitRsp2Client.Magnify[factorType] = serverLimit.Magnify + + log.LogInfof("action[initClientQosInfo] vol [%v] clientID [%v] factorType [%v] init client info and set limitRsp2Client [%v]"+ + "server total[%v] used [%v] buffer [%v]", + qosManager.vol.Name, clientID, proto.QosTypeString(factorType), + initLimit, serverLimit.Total, serverLimit.Allocated, serverLimit.Buffer) + factorType++ + } + + qosManager.cliInfoMgrMap[clientID] = &ClientInfoMgr{ + Cli: clientInitInfo, + Assign: limitRsp2Client, + Time: time.Now(), + ID: clientID, + host: host, + } + log.LogInfof("action[initClientQosInfo] vol [%v] clientID [%v]", qosManager.vol.Name, clientID) + return +} + +func (serverLimit *ServerFactorLimit) getDstLimit(factorType uint32, used, need uint64) (dstLimit uint64) { + if factorType == proto.FlowWriteType || factorType == proto.FlowReadType { + if need > used { + need = used + } + if (need + used) < 10*util.MB/8 { + dstLimit = uint64(float32(need+used) * 2) + } else if (need + used) < 50*util.MB/8 { + dstLimit = uint64(float32(need+used) * 1.5) + } else if (need + used) < 100*util.MB/8 { + dstLimit = uint64(float32(need+used) * 1.2) + } else if (need + used) < 1*util.GB/8 { + dstLimit = uint64(float32(need+used) * 1.1) + } else { + dstLimit = uint64(float32(need+used) + 1*util.GB/8) + } + } else { + if (need + used) < 100 { + dstLimit = uint64(float32(need+used) * 2) + } else if (need + used) < 500 { + dstLimit = uint64(float32(need+used) * 1.5) + } else if (need + used) < 1000 { + dstLimit = uint64(float32(need+used) * 1.2) + } else if (need + used) < 5000 { + dstLimit = uint64(float32(need+used) * 1.2) + } else { + dstLimit = uint64(float32(need+used) + 1000) + } + } + return +} + +func (serverLimit *ServerFactorLimit) dispatch() { + log.LogInfof("action[dispatch] type [%v] start!", serverLimit.Type) + for { + select { + case request := <-serverLimit.requestCh: + serverLimit.updateLimitFactor(request) + case <-serverLimit.done: + log.LogErrorf("done ServerFactorLimit type (%v)", serverLimit.Type) + return + } + } +} + +// handle client request and rsp with much more if buffer is enough according rules of allocate +func (s *ServerFactorLimit) updateLimitFactor(req interface{}) { + + request := req.(*qosRequestArgs) + clientID := request.clientID + factorType := request.factorType + clientReq := request.clientReq + assignInfo := request.assignInfo + rsp2Client := request.rsp2Client + lastClientInfo := request.lastClientInfo + + log.LogInfof("action[updateLimitFactor] vol [%v] clientID [%v] type [%v],client report [%v,%v,%v,%v] last client report [%v,%v,%v,%v] periodically cal Assign [%v,%v]", + s.qosManager.vol.Name, clientID, proto.QosTypeString(factorType), + clientReq.Used, clientReq.Need, clientReq.UsedLimit, clientReq.UsedBuffer, + lastClientInfo.Used, lastClientInfo.Need, lastClientInfo.UsedLimit, lastClientInfo.UsedBuffer, + assignInfo.UsedLimit, assignInfo.UsedBuffer) + + rsp2Client.UsedLimit = assignInfo.UsedLimit + rsp2Client.UsedBuffer = assignInfo.UsedBuffer + + // flow limit and buffer not enough,client need more + if (clientReq.Need + clientReq.Used) > (assignInfo.UsedLimit + assignInfo.UsedBuffer) { + log.LogInfof("action[updateLimitFactor] vol [%v] clientID [%v] type [%v], need [%v] used [%v], used limit [%v]", + s.qosManager.vol.Name, clientID, proto.QosTypeString(factorType), clientReq.Need, clientReq.Used, clientReq.UsedLimit) + + dstLimit := s.getDstLimit(factorType, clientReq.Used, clientReq.Need) + + // Assign already allocated the buffer for client + if dstLimit > assignInfo.UsedLimit+assignInfo.UsedBuffer { + additionBuffer := dstLimit - assignInfo.UsedLimit - assignInfo.UsedBuffer + // if buffer is available then balance must not effect, try use buffer as possible as can + if s.Buffer > 0 { + log.LogDebugf("action[updateLimitFactor] vol [%v] clientID [%v] type [%v] client need more buffer [%v] serverlimit buffer [%v] used [%v]", + s.qosManager.vol.Name, clientID, proto.QosTypeString(factorType), + additionBuffer, s.Buffer, s.Allocated) + + // calc dst buffer for client to expand + // ignore the case of s.used be zero. used should large then 0 because dstLimit isn't zero and be part of s.used + dstUsedBuffer := uint64(float32(dstLimit*s.Buffer/s.Allocated) * 0.5) + if dstUsedBuffer > dstLimit { + dstUsedBuffer = dstLimit + } + if assignInfo.UsedBuffer < dstUsedBuffer { + additionBuffer = dstUsedBuffer - assignInfo.UsedBuffer + if additionBuffer > s.Buffer { + rsp2Client.UsedBuffer += s.Buffer + assignInfo.UsedBuffer = rsp2Client.UsedBuffer + s.Allocated += s.Buffer + s.Buffer = 0 + } else { + rsp2Client.UsedBuffer = dstUsedBuffer + assignInfo.UsedBuffer = dstUsedBuffer + s.Buffer -= additionBuffer + s.Allocated += additionBuffer + } + } + } + } + } + log.LogInfof("action[updateLimitFactor] vol [%v] [clientID [%v] type [%v] rsp2Client.UsedLimit [%v], UsedBuffer [%v]", + s.qosManager.vol.Name, clientID, proto.QosTypeString(factorType), rsp2Client.UsedLimit, rsp2Client.UsedBuffer) + request.wg.Done() +} + +func (qosManager *QosCtrlManager) init(cluster *Cluster, host string) (limit *proto.LimitRsp2Client, err error) { + log.LogInfof("action[qosManage.init] vol [%v] host %v", qosManager.vol.Name, host) + var id uint64 + if id, err = cluster.idAlloc.allocateCommonID(); err == nil { + return qosManager.initClientQosInfo(id, host) + } + return +} + +func (qosManager *QosCtrlManager) HandleClientQosReq(reqClientInfo *proto.ClientReportLimitInfo, clientID uint64) (limitRsp *proto.LimitRsp2Client, err error) { + log.LogInfof("action[HandleClientQosReq] vol [%v] reqClientInfo from [%v], enable [%v]", + qosManager.vol.Name, clientID, qosManager.qosEnable) + + qosManager.RLock() + clientInfo, lastExist := qosManager.cliInfoMgrMap[clientID] + if !lastExist || reqClientInfo == nil { + qosManager.RUnlock() + log.LogWarnf("action[HandleClientQosReq] vol [%v] id [%v] addr [%v] not exist", qosManager.vol.Name, clientID, reqClientInfo.Host) + return qosManager.initClientQosInfo(clientID, reqClientInfo.Host) + } + qosManager.RUnlock() + + limitRsp = proto.NewLimitRsp2Client() + limitRsp.Enable = qosManager.qosEnable + limitRsp.ID = reqClientInfo.ID + limitRsp.ReqPeriod = qosManager.ClientReqPeriod + limitRsp.HitTriggerCnt = uint8(qosManager.ClientHitTriggerCnt) + + if !qosManager.qosEnable { + clientInfo.Cli = reqClientInfo + limitRsp.FactorMap = reqClientInfo.FactorMap + clientInfo.Assign = limitRsp + clientInfo.Time = time.Now() + for i := proto.IopsReadType; i <= proto.FlowWriteType; i++ { + log.LogInfof("action[HandleClientQosReq] vol [%v] [%v,%v,%v,%v]", qosManager.vol.Name, + reqClientInfo.FactorMap[i].Used, + reqClientInfo.FactorMap[i].Need, + reqClientInfo.FactorMap[i].UsedLimit, + reqClientInfo.FactorMap[i].UsedBuffer) + } + return + } + + index := 0 + wg := &sync.WaitGroup{} + wg.Add(len(reqClientInfo.FactorMap)) + for factorType, clientFactor := range reqClientInfo.FactorMap { + limitRsp.FactorMap[factorType] = &proto.ClientLimitInfo{} + serverLimit := qosManager.serverFactorLimitMap[factorType] + limitRsp.Magnify[factorType] = serverLimit.Magnify + + request := &qosRequestArgs{ + clientID: clientID, + factorType: factorType, + clientReq: clientFactor, + lastClientInfo: clientInfo.Cli.FactorMap[factorType], + assignInfo: clientInfo.Assign.FactorMap[factorType], + rsp2Client: limitRsp.FactorMap[factorType], + wg: wg, + } + serverLimit.requestCh <- request + index++ + } + wg.Wait() + + clientInfo.Cli = reqClientInfo + clientInfo.Assign = limitRsp + clientInfo.Time = time.Now() + + return +} +func (qosManager *QosCtrlManager) updateServerLimitByClientsInfo(factorType uint32) { + var ( + cliSum proto.ClientLimitInfo + nextStageNeed, nextStageUse uint64 + ) + qosManager.RLock() + + // get sum of data from all clients reports + for host, cliInfo := range qosManager.cliInfoMgrMap { + cliFactor := cliInfo.Cli.FactorMap[factorType] + cliSum.Used += cliFactor.Used + cliSum.Need += cliFactor.Need + cliSum.UsedLimit += cliFactor.UsedLimit + cliSum.UsedBuffer += cliFactor.UsedBuffer + log.LogInfof("action[updateServerLimitByClientsInfo] vol [%v] host [%v] type [%v] used [%v] need [%v] limit [%v] buffer [%v]", + qosManager.vol.Name, host, proto.QosTypeString(factorType), + cliFactor.Used, cliFactor.Need, cliFactor.UsedLimit, cliFactor.UsedBuffer) + } + + log.LogInfof("action[updateServerLimitByClientsInfo] vol [%v] type [%v] all clisum used:[%v] need:[%v] limit:[%v] buffer:[%v]", + qosManager.vol.Name, proto.QosTypeString(factorType), cliSum.Used, cliSum.Need, cliSum.UsedLimit, cliSum.UsedBuffer) + + serverLimit := qosManager.serverFactorLimitMap[factorType] + serverLimit.CliUsed = cliSum.Used + serverLimit.CliNeed = cliSum.Need + qosManager.RUnlock() + + if cliSum.Used > serverLimit.Total { + log.LogWarnf("action[updateServerLimitByClientsInfo] vol [%v] type [%v] sum of all clients flow use [%v] larger than volume totoal [%v]", + qosManager.vol.Name, proto.QosTypeString(factorType), cliSum.Used, serverLimit.Total) + } + + log.LogInfof("action[updateServerLimitByClientsInfo] vol [%v] type [%v] last epoch serverLimit total:[%v] used:[%v] need:[%v] buffer:[%v] limitrate:[%v]", + qosManager.vol.Name, proto.QosTypeString(factorType), serverLimit.Total, serverLimit.Allocated, serverLimit.NeedAfterAlloc, serverLimit.Buffer, serverLimit.LimitRate) + + serverLimit.Buffer = 0 + nextStageUse = cliSum.Used + nextStageNeed = cliSum.Need + if serverLimit.Total >= nextStageUse { + serverLimit.Buffer = serverLimit.Total - nextStageUse + log.LogInfof("action[updateServerLimitByClientsInfo] vol [%v] reset server buffer [%v] all clients nextStageUse [%v]", + qosManager.vol.Name, serverLimit.Buffer, nextStageUse) + if nextStageNeed > serverLimit.Buffer { + nextStageNeed -= serverLimit.Buffer + nextStageUse += serverLimit.Buffer + serverLimit.Buffer = 0 + log.LogInfof("action[updateServerLimitByClientsInfo] vol [%v] reset server buffer [%v] all clients nextStageNeed [%v] too nuch", + qosManager.vol.Name, serverLimit.Buffer, nextStageNeed) + } else { + serverLimit.Buffer -= nextStageNeed + log.LogInfof("action[updateServerLimitByClientsInfo] vol [%v] reset server buffer [%v] all clients nextStageNeed [%v]", + qosManager.vol.Name, serverLimit.Buffer, nextStageNeed) + nextStageUse += nextStageNeed + nextStageNeed = 0 + } + } else { // usage large than limitation + log.LogInfof("action[updateServerLimitByClientsInfo] vol[%v] type [%v] clients needs [%v] plus overuse [%v],get nextStageNeed [%v]", + qosManager.vol.Name, proto.QosTypeString(factorType), nextStageNeed, nextStageUse-serverLimit.Total, + nextStageNeed+nextStageUse-serverLimit.Total) + nextStageNeed += nextStageUse - serverLimit.Total + nextStageUse = serverLimit.Total + } + + serverLimit.Allocated = nextStageUse + serverLimit.NeedAfterAlloc = nextStageNeed + + log.LogInfof("action[updateServerLimitByClientsInfo] vol [%v] type [%v] after cal get next stage need [%v] and used [%v], buffer [%v]", + qosManager.vol.Name, proto.QosTypeString(factorType), nextStageNeed, nextStageUse, serverLimit.Buffer) + // get the limitRate,additionFlowNeed should be zero if total used can increase + serverLimit.LimitRate = 0 + if serverLimit.NeedAfterAlloc > 0 { + serverLimit.LimitRate = float32(serverLimit.NeedAfterAlloc) / float32(serverLimit.Allocated+serverLimit.NeedAfterAlloc) + } + + return +} + +func (qosManager *QosCtrlManager) assignClientsNewQos(factorType uint32) { + qosManager.RLock() + serverLimit := qosManager.serverFactorLimitMap[factorType] + var bufferAllocated uint64 + // recalculate client Assign limit and buffer + for host, cliInfoMgr := range qosManager.cliInfoMgrMap { + cliInfo := cliInfoMgr.Cli.FactorMap[factorType] + assignInfo := cliInfoMgr.Assign.FactorMap[factorType] + + if cliInfo.Used+cliInfoMgr.Cli.FactorMap[factorType].Need == 0 { + assignInfo.UsedLimit = 0 + assignInfo.UsedBuffer = 0 + } else { + assignInfo.UsedLimit = uint64(float32(cliInfo.Used+cliInfo.Need) * (1 - serverLimit.LimitRate)) + if serverLimit.Allocated != 0 { + assignInfo.UsedBuffer = uint64(float32(serverLimit.Buffer*assignInfo.UsedLimit/serverLimit.Allocated) * 0.5) + } + + log.LogDebugf("action[assignClientsNewQos] Assign host [%v] limit [%v] buffer [%v]", + host, assignInfo.UsedLimit, assignInfo.UsedBuffer) + + // buffer left may be quit large and we should not used up and doen't mean if buffer large than used limit line + if assignInfo.UsedBuffer > assignInfo.UsedLimit { + assignInfo.UsedBuffer = assignInfo.UsedLimit + } + } + + bufferAllocated += assignInfo.UsedBuffer + log.LogInfof("action[assignClientsNewQos] vol [%v] host [%v] type [%v] assignInfo used limit [%v], used buffer [%v]", + qosManager.vol.Name, host, proto.QosTypeString(factorType), assignInfo.UsedLimit, assignInfo.UsedBuffer) + } + + qosManager.RUnlock() + + if serverLimit.Buffer > bufferAllocated { + serverLimit.Buffer -= bufferAllocated + } else { + serverLimit.Buffer = 0 + log.LogWarnf("action[assignClientsNewQos] vol [%v] type [%v] clients buffer [%v] and server buffer used up trigger flow limit overall", + qosManager.vol.Name, proto.QosTypeString(factorType), bufferAllocated) + } + + log.LogInfof("action[assignClientsNewQos] vol [%v] type [%v] serverLimit buffer:[%v] used:[%v] need:[%v] total:[%v]", + qosManager.vol.Name, proto.QosTypeString(factorType), + serverLimit.Buffer, serverLimit.Allocated, serverLimit.NeedAfterAlloc, serverLimit.Total) +} + +func (vol *Vol) checkQos() { + vol.qosManager.Lock() + // check expire client and delete from map + tTime := time.Now() + for id, cli := range vol.qosManager.cliInfoMgrMap { + if cli.Time.Add(20 * time.Second).Before(tTime) { + log.LogWarnf("action[checkQos] vol [%v] Id [%v] addr [%v] be delete in case of long time no request", + vol.Name, id, cli.host) + delete(vol.qosManager.cliInfoMgrMap, id) + } + } + + if !vol.qosManager.qosEnable { + vol.qosManager.Unlock() + return + } + vol.qosManager.Unlock() + + // periodically updateServerLimitByClientsInfo and get assigned limit info for all clients + // with last report info from client and qos control info + for factorType := proto.IopsReadType; factorType <= proto.FlowWriteType; factorType++ { + // calc all clients and get real used and need value , used value should less then total + vol.qosManager.updateServerLimitByClientsInfo(factorType) + // update client assign info by result above + vol.qosManager.assignClientsNewQos(factorType) + + serverLimit := vol.qosManager.serverFactorLimitMap[factorType] + log.LogInfof("action[UpdateAllQosInfo] vol name [%v] type [%v] after updateServerLimitByClientsInfo get limitRate:[%v] server total [%v] used [%v] need [%v] buffer [%v]", + vol.Name, proto.QosTypeString(factorType), serverLimit.LimitRate, serverLimit.Total, serverLimit.Allocated, serverLimit.NeedAfterAlloc, serverLimit.Buffer) + } +} + +func (vol *Vol) getQosStatus() interface{} { + + type qosStatus struct { + ServerFactorLimitMap map[uint32]*ServerFactorLimit // vol qos data for iops w/r and flow w/r + DefaultClientCnt uint32 + InitClientCnt uint32 + QosEnable bool + ClientReqPeriod uint32 + ClientHitTriggerCnt uint32 + } + + return &qosStatus{ + ServerFactorLimitMap: vol.qosManager.serverFactorLimitMap, + QosEnable: vol.qosManager.qosEnable, + ClientReqPeriod: vol.qosManager.ClientReqPeriod, + ClientHitTriggerCnt: vol.qosManager.ClientHitTriggerCnt, + } +} + +func (vol *Vol) getClientLimitInfo(id uint64, ip string) (interface{}, error) { + log.LogInfof("action[getClientLimitInfo] vol [%v] id [%v] ip [%v]", vol.Name, id, ip) + vol.qosManager.RLock() + defer vol.qosManager.RUnlock() + + if id > 0 { + if info, ok := vol.qosManager.cliInfoMgrMap[id]; ok { + if len(ip) > 0 && util.GetIp(info.host) != ip { + return nil, fmt.Errorf("ip info [%v] not equal with request [%v]", info.host, ip) + } + return info, nil + } + } else { + if len(ip) != 0 { + var resp []*ClientInfoMgr + for _, info := range vol.qosManager.cliInfoMgrMap { + // http connection port from client will change time by time,so ignore port here + if util.GetIp(info.host) == ip { + resp = append(resp, info) + } + } + if len(resp) > 0 { + return resp, nil + } + } else { + return vol.qosManager.cliInfoMgrMap, nil + } + } + return nil, fmt.Errorf("not found") +} + +func (vol *Vol) volQosEnable(c *Cluster, enable bool) error { + log.LogWarnf("action[qosEnable] vol %v, set qos enable [%v], qosmgr[%v]", vol.Name, enable, vol.qosManager) + vol.qosManager.qosEnable = enable + return c.syncUpdateVol(vol) +} + +func (vol *Vol) updateClientParam(c *Cluster, period, triggerCnt uint64) error { + vol.qosManager.ClientHitTriggerCnt = uint32(triggerCnt) + vol.qosManager.ClientReqPeriod = uint32(period) + return c.syncUpdateVol(vol) +} + +func (vol *Vol) volQosUpdateMagnify(c *Cluster, magnifyArgs *qosArgs) error { + vol.qosManager.volUpdateMagnify(magnifyArgs) + return c.syncUpdateVol(vol) +} + +func (vol *Vol) volQosUpdateLimit(c *Cluster, limitArgs *qosArgs) error { + vol.qosManager.volUpdateLimit(limitArgs) + return c.syncUpdateVol(vol) +} + func (vol *Vol) updateViewCache(c *Cluster) { view := proto.NewVolView(vol.Name, vol.Status, vol.FollowerRead, vol.createTime, vol.CacheTTL, vol.VolType) view.SetOwner(vol.Owner) diff --git a/objectnode/fs_volume.go b/objectnode/fs_volume.go index df7b5e3f9..88ecad912 100644 --- a/objectnode/fs_volume.go +++ b/objectnode/fs_volume.go @@ -49,7 +49,7 @@ const ( // These asynchronous tasks include periodic volume topology and metadata update tasks. type AsyncTaskErrorFunc func(err error) -// OnError protects the call of AsyncTaskErrorFunc with null pointer access. Used to simplify caller code. +// OnError protects the call of AsyncTaskErrorFunc with null pointer access. Allocated to simplify caller code. func (f AsyncTaskErrorFunc) OnError(err error) { if f != nil { f(err) diff --git a/objectnode/util.go b/objectnode/util.go index a94e3f6dd..f0de3e9cb 100644 --- a/objectnode/util.go +++ b/objectnode/util.go @@ -53,7 +53,7 @@ type PathItem struct { IsDirectory bool } -// PathIterator is a path iterator. Used to sequentially iterate each path node from a complete path. +// PathIterator is a path iterator. Allocated to sequentially iterate each path node from a complete path. type PathIterator struct { cursor int path string diff --git a/proto/admin_proto.go b/proto/admin_proto.go index bfac5d4ee..92255c5c4 100644 --- a/proto/admin_proto.go +++ b/proto/admin_proto.go @@ -69,6 +69,15 @@ const ( ClientVolStat = "/client/volStat" ClientMetaPartitions = "/client/metaPartitions" + // qos api + QosGetStatus = "/qos/getStatus" + QosGetClientsLimitInfo = "/qos/getClientsInfo" + QosGetZoneLimitInfo = "/qos/getZoneLimit" // include disk enable + QosUpdate = "/qos/update" // include disk enable + QosUpdateMagnify = "/qos/updateMagnify" + QosUpdateClientParam = "/qos/updateClientParam" + QosUpdateZoneLimit = "/qos/updateZoneLimit" // include disk enable + QosUpload = "/admin/qosUpload" //raft node APIs AddRaftNode = "/raftNode/add" RemoveRaftNode = "/raftNode/remove" @@ -257,10 +266,19 @@ type LoadMetaPartitionMetricResponse struct { Result string } +type QosToDataNode struct { + EnableDiskQos bool + QosIopsReadLimit uint64 + QosIopsWriteLimit uint64 + QosFlowReadLimit uint64 + QosFlowWriteLimit uint64 +} + // HeartBeatRequest define the heartbeat request. type HeartBeatRequest struct { CurrTime int64 MasterAddr string + QosToDataNode } // PartitionReport defines the partition report. @@ -276,6 +294,15 @@ type PartitionReport struct { NeedCompare bool } +type DataNodeQosResponse struct { + IopsRLimit uint64 + IopsWLimit uint64 + FlowRlimit uint64 + FlowWlimit uint64 + Status uint8 + Result string +} + // DataNodeHeartbeatResponse defines the response to the data node heartbeat. type DataNodeHeartbeatResponse struct { Total uint64 @@ -481,6 +508,79 @@ func NewMetaPartitionView(partitionID, start, end uint64, status int8) (mpView * return } +const ( + QosStateNormal uint8 = 0x01 + QosStateHitLimit uint8 = 0x02 +) + +const ( + IopsReadType uint32 = 0x01 + IopsWriteType uint32 = 0x02 + FlowReadType uint32 = 0x03 + FlowWriteType uint32 = 0x04 +) + +const ( + QosDefaultClientCnt uint32 = 100 + QosDefaultDiskMaxFLowLimit int = 0x7FFFFFFF + QosDefaultDiskMaxIoLimit int = 100000 +) + +func QosTypeString(factorType uint32) string { + switch factorType { + case IopsReadType: + return "IopsRead" + case IopsWriteType: + return "IopsWrite" + case FlowReadType: + return "FlowRead" + case FlowWriteType: + return "FlowWrite" + } + return "unkown" +} + +type ClientLimitInfo struct { + UsedLimit uint64 + UsedBuffer uint64 + Used uint64 + Need uint64 +} + +type ClientReportLimitInfo struct { + Ver int8 + ID uint64 + FactorMap map[uint32]*ClientLimitInfo + Host string + Status uint8 + Reserved string +} + +func NewClientReportLimitInfo() *ClientReportLimitInfo { + return &ClientReportLimitInfo{ + FactorMap: make(map[uint32]*ClientLimitInfo, 0), + } +} + +type LimitRsp2Client struct { + Ver int8 + ID uint64 + Enable bool + ReqPeriod uint32 + HitTriggerCnt uint8 + FactorMap map[uint32]*ClientLimitInfo + Magnify map[uint32]uint32 + Reserved string +} + +func NewLimitRsp2Client() *LimitRsp2Client { + limit := &LimitRsp2Client{ + FactorMap: make(map[uint32]*ClientLimitInfo, 0), + Magnify: make(map[uint32]uint32, 0), + } + return limit +} + // SimpleVolView defines the simple view of a volume type SimpleVolView struct { ID uint64 diff --git a/proto/fs_proto.go b/proto/fs_proto.go index 852cbef16..461b2fbb5 100644 --- a/proto/fs_proto.go +++ b/proto/fs_proto.go @@ -29,6 +29,7 @@ const ( const ( FlagsSyncWrite int = 1 << iota FlagsAppend + FlagsCache ) // Mode returns the fileMode. diff --git a/proto/packet.go b/proto/packet.go index aa504c852..ad342e151 100644 --- a/proto/packet.go +++ b/proto/packet.go @@ -123,6 +123,7 @@ const ( OpAddDataPartitionRaftMember uint8 = 0x67 OpRemoveDataPartitionRaftMember uint8 = 0x68 OpDataPartitionTryToLeader uint8 = 0x69 + OpQos uint8 = 0x6A // Operations: MultipartInfo OpCreateMultipart uint8 = 0x70 diff --git a/regression/overlapping/main.go b/regression/overlapping/main.go index e8c198ac5..a8133d6f2 100644 --- a/regression/overlapping/main.go +++ b/regression/overlapping/main.go @@ -43,7 +43,7 @@ func main() { } defer file2.Close() - // Need to write to normal extent, so avoid writing to the beginning 1M + // NeedAfterAlloc to write to normal extent, so avoid writing to the beginning 1M file1.Seek(1*1024*1024, 0) file2.Seek(1*1024*1024, 0) diff --git a/sdk/data/blobstore/reader.go b/sdk/data/blobstore/reader.go index f3e627902..da573e3f4 100644 --- a/sdk/data/blobstore/reader.go +++ b/sdk/data/blobstore/reader.go @@ -17,6 +17,7 @@ package blobstore import ( "context" "fmt" + "github.com/cubefs/cubefs/sdk/data/manager" "io" "os" "sync" @@ -82,6 +83,7 @@ type Reader struct { fileLength uint64 valid bool inflightL2cache sync.Map + limitManager *manager.LimitManager } type ClientConfig struct { @@ -122,6 +124,7 @@ func NewReader(config ClientConfig) (reader *Reader) { reader.ec.UpdateDataPartitionForColdVolume() } + reader.limitManager = reader.ec.LimitManager return } @@ -331,6 +334,7 @@ func (reader *Reader) readSliceRange(ctx context.Context, rs *rwSlice) (err erro log.LogDebugf("TRACE blobStore readSliceRange. cfs block miss.extentKey=%v,err=%v", rs.extentKey, err) } + reader.limitManager.ReadAlloc(ctx, int(rs.rSize)) readN, err = reader.ebs.Read(ctx, reader.volName, buf, rs.rOffset, uint64(rs.rSize), rs.objExtentKey) if err != nil { reader.err <- err @@ -369,6 +373,9 @@ func (reader *Reader) asyncCache(ctx context.Context, cacheKey string, objExtent reader.inflightL2cache.Store(cacheKey, true) defer reader.inflightL2cache.Delete(cacheKey) + // qos control + reader.limitManager.ReadAlloc(ctx, int(objExtentKey.Size)) + buf := make([]byte, objExtentKey.Size) read, err := reader.ebs.Read(ctx, reader.volName, buf, 0, uint64(len(buf)), objExtentKey) if err != nil || read != len(buf) { @@ -378,7 +385,7 @@ func (reader *Reader) asyncCache(ctx context.Context, cacheKey string, objExtent } if reader.needCacheL2() { - reader.ec.Write(reader.ino, int(objExtentKey.FileOffset), buf, 0) + reader.ec.Write(reader.ino, int(objExtentKey.FileOffset), buf, proto.FlagsCache) log.LogDebugf("TRACE blobStore asyncCache(L2) Exit. cacheKey=%v", cacheKey) return } diff --git a/sdk/data/blobstore/writer.go b/sdk/data/blobstore/writer.go index 926d38c5c..9807f5501 100644 --- a/sdk/data/blobstore/writer.go +++ b/sdk/data/blobstore/writer.go @@ -17,6 +17,7 @@ package blobstore import ( "context" "fmt" + "github.com/cubefs/cubefs/sdk/data/manager" "github.com/cubefs/cubefs/util/buf" "sync" "sync/atomic" @@ -59,6 +60,7 @@ type Writer struct { cacheThreshold int dirty bool blockPosition int + limitManager *manager.LimitManager } func NewWriter(config ClientConfig) (writer *Writer) { @@ -84,6 +86,8 @@ func NewWriter(config ClientConfig) (writer *Writer) { writer.cacheThreshold = config.CacheThreshold writer.dirty = false writer.AllocateCache() + writer.limitManager = writer.ec.LimitManager + return } @@ -257,6 +261,7 @@ func (writer *Writer) writeSlice(ctx context.Context, wSlice *rwSlice, wg bool) if wg { defer writer.wg.Done() } + writer.limitManager.WriteAlloc(ctx, int(wSlice.size)) log.LogDebugf("TRACE blobStore,writeSlice to ebs. ino(%v) fileOffset(%v) len(%v)", writer.ino, wSlice.fileOffset, wSlice.size) location, err := writer.ebsc.Write(ctx, writer.volName, wSlice.Data, wSlice.size) if err != nil { @@ -301,7 +306,7 @@ func (writer *Writer) asyncCache(ino uint64, offset int, data []byte) { }() log.LogDebugf("TRACE asyncCache Enter,fileOffset(%v) len(%v)", offset, len(data)) - write, err := writer.ec.Write(ino, offset, data, 0) + write, err := writer.ec.Write(ino, offset, data, proto.FlagsCache) log.LogDebugf("TRACE asyncCache Exit,write(%v) err(%v)", write, err) } diff --git a/sdk/data/manager/limiter.go b/sdk/data/manager/limiter.go new file mode 100644 index 000000000..7316219f5 --- /dev/null +++ b/sdk/data/manager/limiter.go @@ -0,0 +1,476 @@ +package manager + +import ( + "container/list" + "golang.org/x/net/context" + "sync" + "sync/atomic" + "time" + + "github.com/cubefs/cubefs/proto" + "github.com/cubefs/cubefs/sdk/data/wrapper" + "github.com/cubefs/cubefs/util" + "github.com/cubefs/cubefs/util/log" +) + +const ( + runNow = 1 + runLater = 2 + gridHitLimitCnt = 1 + girdCntOneSecond = 3 + gridWindowTimeScope = 10 + qosExpireTime = 20 + qosReportMinGap = 500 + defaultMagnifyFactor = 100 +) + +type UploadFlowInfoFunc func(clientInfo wrapper.SimpleClientInfo) error + +type GridElement struct { + time time.Time + used uint64 + limit uint64 + buffer uint64 + hitLimit bool + ID uint64 + sync.RWMutex +} + +type AllocElement struct { + used uint32 + magnify uint32 + future *util.Future +} + +type LimitFactor struct { + factorType uint32 + gridList *list.List + waitList *list.List + need uint64 + gidHitLimitCnt uint8 + mgr *LimitManager + gridId uint64 + magnify uint32 + lock sync.RWMutex +} + +func (factor *LimitFactor) getNeedByMagnify(allocCnt uint32, magnify uint32) uint64 { + if magnify == 0 { + return 0 + } + if allocCnt > 1000 { + log.LogInfof("action[getNeedByMagnify] allocCnt %v", allocCnt) + magnify = defaultMagnifyFactor + } + + need := uint64(allocCnt * magnify) + if factor.factorType == proto.FlowWriteType || factor.factorType == proto.FlowReadType { + if need > util.GB/8 { + need = util.GB / 8 + } + } + return need +} + +func (factor *LimitFactor) alloc(allocCnt uint32) (ret uint8,future *util.Future) { + //log.LogInfof("action[alloc] type [%v] alloc [%v], tmp factor waitlist [%v] limtcnt [%v] need [%v] len [%v]", proto.QosTypeString(factor.factorType), + // allocCnt, factor.waitList.Len(), factor.gidHitLimitCnt, factor.need, factor.gridList.Len()) + + if !factor.mgr.enable { + // used not accurate also fine, the purpose is get master's info + // without lock can better performance just the used value large than 0 + gridEnd := factor.gridList.Back() + if gridEnd != nil { + grid := gridEnd.Value.(*GridElement) + grid.used = grid.used+uint64(allocCnt) + } + return runNow, nil + } + + type activeSt struct { + activeUpdate bool + needWait bool + } + activeState := &activeSt{} + defer func(active *activeSt) { + if !active.needWait { + factor.lock.RUnlock() + } else if !active.activeUpdate { + factor.lock.Unlock() + } + + }(activeState) + + factor.lock.RLock() + grid := factor.gridList.Back().Value.(*GridElement) + + if factor.mgr.enable && (factor.waitList.Len() > 0 || grid.used+uint64(allocCnt) > grid.limit+grid.buffer) { + factor.lock.RUnlock() + factor.lock.Lock() + activeState.needWait = true + future = util.NewFuture() + + factor.waitList.PushBack(&AllocElement{ + used: allocCnt, + future: future, + magnify: factor.magnify, + }) + + factor.need += factor.getNeedByMagnify(allocCnt, factor.magnify) + + if grid.hitLimit == false { + factor.gidHitLimitCnt++ + // 1s have several gird, gidHitLimitCnt is the count that gird count hit limit in latest 1s, + // if gidHitLimitCnt large than limit then request for enlarge factor limit + // GetSimpleVolView will call back simpleClient function to get factor info and send to master + if factor.gidHitLimitCnt >= factor.mgr.HitTriggerCnt { + tmpTime := time.Now() + if factor.mgr.lastReqTime.Add(time.Duration(factor.mgr.ReqPeriod) * time.Second).Before(tmpTime) { + factor.mgr.lastReqTime = tmpTime + log.LogInfof("CheckGrid factor [%v] unlock before active update simple vol view,gird id[%v] limit[%v] buffer [%v] used [%v]", + proto.QosTypeString(factor.factorType), grid.ID, grid.limit, grid.buffer, grid.used) + // unlock need call here,UpdateSimpleVolView will lock again + factor.lock.Unlock() + activeState.activeUpdate = true + go factor.mgr.WrapperUpdate(factor.mgr.simpleClient) + } + } + } + grid.hitLimit = true + return runLater, future + } + + atomic.CompareAndSwapUint64(&grid.used, grid.used, grid.used+uint64(allocCnt)) + return runNow, future +} + +func (factor *LimitFactor) SetLimit(limitVal uint64, bufferVal uint64) { + log.LogInfof("acton[SetLimit] factor type [%v] limitVal [%v] bufferVal [%v]", proto.QosTypeString(factor.factorType), limitVal, bufferVal) + var grid *GridElement + factor.mgr.lastTimeOfSetLimit = time.Now() + factor.lock.Lock() + + defer func() { + factor.TryReleaseWaitList(grid) + factor.lock.Unlock() + }() + + if factor.gridList.Len() == 0 { + grid = &GridElement{ + time: time.Now(), + limit: limitVal / girdCntOneSecond, + buffer: bufferVal / girdCntOneSecond, + ID: factor.gridId, + } + factor.gridId++ + factor.gridList.PushBack(grid) + } else { + grid = factor.gridList.Back().Value.(*GridElement) + grid.buffer = bufferVal / girdCntOneSecond + grid.limit = limitVal / girdCntOneSecond + } +} + +// clean wait list if limit be enlarged by master +// no lock need for parallel,caller own the lock and will release it +func (factor *LimitFactor) TryReleaseWaitList(newGrid *GridElement) { + + for factor.waitList.Len() > 0 { + value := factor.waitList.Front() + ele := value.Value.(*AllocElement) + if newGrid.used+uint64(ele.used) > newGrid.limit+newGrid.buffer { + log.LogWarnf("action[TryReleaseWaitList] type [%v] new gird be used up.alloc in waitlist left cnt [%v],"+ + "grid be allocated [%v] grid limit [%v] and buffer[%v], gird id:[%v]", proto.QosTypeString(factor.factorType), + factor.waitList.Len(), newGrid.used, newGrid.limit, newGrid.buffer, newGrid.ID) + break + } + newGrid.used += uint64(ele.used) + ele.future.Respond(true, nil) + + factor.need -= factor.getNeedByMagnify(ele.used, ele.magnify) + value = value.Next() + factor.waitList.Remove(factor.waitList.Front()) + } +} + +func (factor *LimitFactor) CheckGrid() { + defer func() { + factor.lock.Unlock() + }() + factor.lock.Lock() + + grid := factor.gridList.Back().Value.(*GridElement) + newGrid := &GridElement{ + time: time.Now(), + limit: grid.limit, + used: 0, + buffer: grid.buffer, + ID: factor.gridId, + } + factor.gridId++ + + if factor.mgr.enable == true && factor.mgr.lastTimeOfSetLimit.Add(time.Second*qosExpireTime).Before(newGrid.time) { + log.LogWarnf("action[CheckGrid]. qos disable in case of recv no command from master in long time") + } + log.LogInfof("action[CheckGrid] factor type:[%v] gridlistLen:[%v] waitlistLen:[%v] need:[%v] hitlimitcnt:[%v] "+ + "add new grid info used:[%v] limit:[%v] buffer:[%v] time:[%v]", + proto.QosTypeString(factor.factorType), factor.gridList.Len(), factor.waitList.Len(), factor.need, factor.gidHitLimitCnt, + newGrid.used, newGrid.limit, newGrid.buffer, newGrid.time) + + factor.gridList.PushBack(newGrid) + for factor.gridList.Len() > gridWindowTimeScope*girdCntOneSecond { + firstGrid := factor.gridList.Front().Value.(*GridElement) + if firstGrid.hitLimit { + factor.gidHitLimitCnt-- + log.LogInfof("action[CheckGrid] factor [%v] after minus gidHitLimitCnt:[%v]", + factor.factorType, factor.gidHitLimitCnt) + } + log.LogInfof("action[CheckGrid] type:[%v] remove oldest grid info buffer:[%v] limit:[%v] used[%v] from gridlist", + factor.factorType, firstGrid.buffer, firstGrid.limit, firstGrid.used) + factor.gridList.Remove(factor.gridList.Front()) + } + factor.TryReleaseWaitList(newGrid) + +} + +func newLimitFactor(mgr *LimitManager, factorType uint32) *LimitFactor { + limit := &LimitFactor{ + mgr: mgr, + factorType: factorType, + waitList: list.New(), + gridList: list.New(), + magnify: defaultMagnifyFactor, + } + + limit.SetLimit(0, 0) + return limit +} + +type LimitManager struct { + ID uint64 + limitMap map[uint32]*LimitFactor + enable bool + simpleClient wrapper.SimpleClientInfo + exitCh chan struct{} + WrapperUpdate UploadFlowInfoFunc + ReqPeriod uint32 + HitTriggerCnt uint8 + lastReqTime time.Time + lastTimeOfSetLimit time.Time + isLastReqValid bool + once sync.Once +} + +func NewLimitManager(client wrapper.SimpleClientInfo) *LimitManager { + mgr := &LimitManager{ + limitMap: make(map[uint32]*LimitFactor, 0), + enable: false, // assign from master + simpleClient: client, + HitTriggerCnt: gridHitLimitCnt, + ReqPeriod: 1, + } + mgr.limitMap[proto.IopsReadType] = newLimitFactor(mgr, proto.IopsReadType) + mgr.limitMap[proto.IopsWriteType] = newLimitFactor(mgr, proto.IopsWriteType) + mgr.limitMap[proto.FlowWriteType] = newLimitFactor(mgr, proto.FlowWriteType) + mgr.limitMap[proto.FlowReadType] = newLimitFactor(mgr, proto.FlowReadType) + + mgr.ScheduleCheckGrid() + return mgr +} + +func (limitManager *LimitManager) GetFlowInfo() (*proto.ClientReportLimitInfo, bool) { + log.LogInfof("action[LimitManager.GetFlowInfo]") + info := &proto.ClientReportLimitInfo{ + FactorMap: make(map[uint32]*proto.ClientLimitInfo, 0), + } + var validCliInfo bool + for factorType, limitFactor := range limitManager.limitMap { + limitFactor.lock.RLock() + + var used, limit, buffer uint64 + grid := limitFactor.gridList.Back() + // calc and set in case of time may be shifted + griCnt := 0 + if limitFactor.gridList.Len() > 0 { + log.LogInfof("action[GetFlowInfo] start grid %v", grid.Value.(*GridElement).ID) + } + + for griCnt < limitFactor.gridList.Len() { + used += grid.Value.(*GridElement).used + limit += grid.Value.(*GridElement).limit + buffer += grid.Value.(*GridElement).buffer + griCnt++ + if grid.Prev() == nil || griCnt >= girdCntOneSecond { + break + } + + grid = grid.Prev() + } + if limitFactor.gridList.Len() > 0 { + log.LogInfof("action[GetFlowInfo] end grid %v", grid.Value.(*GridElement).ID) + } + + timeElapse := time.Since(grid.Value.(*GridElement).time).Milliseconds() + if timeElapse < qosReportMinGap { + log.LogWarnf("action[GetFlowInfo] timeElapse [%v] since last report", timeElapse) + timeElapse = qosReportMinGap // time of interval get vol view from master todo:change to config time + } + + reqUsed := uint64(float64(used) / (float64(timeElapse) / 1000)) + if (limitFactor.factorType == proto.FlowReadType || limitFactor.factorType == proto.FlowWriteType) && + limitFactor.need > 0 { + if reqUsed < util.MB { + limitFactor.need = 5 * reqUsed + } else if reqUsed < 5*util.MB { + limitFactor.need = 2 * reqUsed + } else if reqUsed < 10*util.MB { + limitFactor.need = uint64(1.5 * float64(reqUsed)) + } else if reqUsed < 50*util.MB { + limitFactor.need = uint64(1.2 * float64(reqUsed)) + } else if reqUsed < 100*util.MB { + limitFactor.need = uint64(1.1 * float64(reqUsed)) + } else if reqUsed < 300*util.MB { + limitFactor.need = reqUsed + } else{ + limitFactor.need = 300 * util.MB + } + } + + factor := &proto.ClientLimitInfo{ + Used: reqUsed, + Need: limitFactor.need, + UsedLimit: limitFactor.gridList.Back().Value.(*GridElement).limit * girdCntOneSecond, + UsedBuffer: limitFactor.gridList.Back().Value.(*GridElement).buffer * girdCntOneSecond, + } + + //if factor.Allocated > 100 && factor.NeedAfterAlloc > factor.Allocated { + // factor.NeedAfterAlloc = factor.Allocated + //} + + limitFactor.lock.RUnlock() + + info.FactorMap[factorType] = factor + info.Host = wrapper.LocalIP + info.Status = proto.QosStateNormal + info.ID = limitManager.ID + if factor.Used|factor.Need|factor.UsedLimit|factor.UsedBuffer > 0 { + validCliInfo = true + } + log.LogInfof("action[GetFlowInfo] type [%v] report to master with simpleClient limit info [%v,%v,%v,%v],host [%v], status [%v] grid [%v, %v, %v]", + proto.QosTypeString(limitFactor.factorType), + factor.Used, factor.Need, factor.UsedBuffer, factor.UsedLimit, + info.Host, info.Status, + grid.Value.(*GridElement).ID, + grid.Value.(*GridElement).limit, + grid.Value.(*GridElement).buffer) + } + + lastValid := limitManager.isLastReqValid + limitManager.isLastReqValid = validCliInfo + + limitManager.once.Do(func() { + validCliInfo = true + }) + // client has no user request then don't report to master + if !lastValid && !validCliInfo { + return info, false + } + return info, true +} + +func (limitManager *LimitManager) ScheduleCheckGrid() { + go func() { + ticker := time.NewTicker(1000 / girdCntOneSecond * time.Millisecond) + defer func() { + ticker.Stop() + }() + + for { + select { + case <-limitManager.exitCh: + return + case <-ticker.C: + for _, limitFactor := range limitManager.limitMap { + limitFactor.CheckGrid() + } + } + } + }() +} + +func (limitManager *LimitManager) SetClientLimit(limit *proto.LimitRsp2Client) { + if limit == nil { + log.LogInfof("action[SetClientLimit] limit info is nil") + return + } + log.LogInfof("action[SetClientLimit] limit enable %v", limit.Enable) + if limitManager.enable != limit.Enable { + log.LogWarnf("action[SetClientLimit] enable [%v]", limit.Enable) + } + limitManager.enable = limit.Enable + if limit.HitTriggerCnt > 0 { + log.LogWarnf("action[SetClientLimit] update to HitTriggerCnt [%v] from [%v]", limitManager.HitTriggerCnt, limit.HitTriggerCnt) + limitManager.HitTriggerCnt = limit.HitTriggerCnt + } + if limit.ReqPeriod > 0 { + log.LogWarnf("action[SetClientLimit] update to ReqPeriod [%v] from [%v]", limitManager.ReqPeriod, limit.ReqPeriod) + limitManager.ReqPeriod = limit.ReqPeriod + } + + for factorType, clientLimitInfo := range limit.FactorMap { + limitManager.limitMap[factorType].SetLimit(clientLimitInfo.UsedLimit, clientLimitInfo.UsedBuffer) + } + for factorType, magnify := range limit.Magnify { + if magnify > 0 && magnify != limitManager.limitMap[factorType].magnify { + log.LogInfof("action[SetClientLimit] type [%v] update magnify [%v] to [%v]", + proto.QosTypeString(factorType), limitManager.limitMap[factorType].magnify, magnify) + limitManager.limitMap[factorType].magnify = magnify + } + } +} + +func (limitManager *LimitManager) ReadAlloc(ctx context.Context, size int) { + limitManager.WaitN(ctx, limitManager.limitMap[proto.IopsReadType], 1) + limitManager.WaitN(ctx, limitManager.limitMap[proto.FlowReadType], size) +} + +func (limitManager *LimitManager) WriteAlloc(ctx context.Context, size int) { + limitManager.WaitN(ctx, limitManager.limitMap[proto.IopsWriteType], 1) + limitManager.WaitN(ctx, limitManager.limitMap[proto.FlowWriteType], size) +} + +// WaitN blocks until alloc success +func (limitManager *LimitManager) WaitN(ctx context.Context, lim *LimitFactor, n int) (err error) { + var fut *util.Future + var ret uint8 + if ret, fut = lim.alloc(uint32(n)); ret == runNow { + // log.LogInfof("action[WaitN] type [%v] return now waitlistlen [%v]", proto.QosTypeString(lim.factorType), lim.waitList.Len()) + return nil + } + + respCh, errCh := fut.AsyncResponse() + + select { + case <-ctx.Done(): + log.LogWarnf("action[WaitN] type [%v] ctx done return waitlistlen [%v]", proto.QosTypeString(lim.factorType), lim.waitList.Len()) + return ctx.Err() + case err = <-errCh: + log.LogWarnf("action[WaitN] type [%v] err return waitlistlen [%v]", proto.QosTypeString(lim.factorType), lim.waitList.Len()) + return + case <-respCh: + log.LogInfof("action[WaitN] type [%v] return waitlistlen [%v]", proto.QosTypeString(lim.factorType), lim.waitList.Len()) + return nil + //default: + } +} + +func (limitManager *LimitManager) UpdateFlowInfo(limit *proto.LimitRsp2Client) { + log.LogInfof("action[LimitManager.UpdateFlowInfo]") + limitManager.SetClientLimit(limit) + return +} + +func (limitManager *LimitManager) SetClientID(id uint64) (err error) { + limitManager.ID = id + return +} diff --git a/sdk/data/stream/extent_client.go b/sdk/data/stream/extent_client.go index fe0a5d9e0..250e8ac6d 100644 --- a/sdk/data/stream/extent_client.go +++ b/sdk/data/stream/extent_client.go @@ -17,6 +17,8 @@ package stream import ( "container/list" "fmt" + "github.com/cubefs/cubefs/sdk/data/manager" + "golang.org/x/time/rate" "sync" "syscall" "time" @@ -131,6 +133,7 @@ type ExtentClient struct { bcacheDir string BcacheHealth bool preload bool + LimitManager *manager.LimitManager dataWrapper *wrapper.Wrapper appendExtentKey AppendExtentKeyFunc getExtents GetExtentsFunc @@ -202,10 +205,11 @@ func (client *ExtentClient) backgroundEvictStream() { // NewExtentClient returns a new extent client. func NewExtentClient(config *ExtentConfig) (client *ExtentClient, err error) { client = new(ExtentClient) - + client.LimitManager = manager.NewLimitManager(client) + client.LimitManager.WrapperUpdate = client.UploadFlowInfo limit := MaxMountRetryLimit retry: - client.dataWrapper, err = wrapper.NewDataPartitionWrapper(config.Volume, config.Masters, config.Preload) + client.dataWrapper, err = wrapper.NewDataPartitionWrapper(client, config.Volume, config.Masters, config.Preload) if err != nil { if limit <= 0 { return nil, errors.Trace(err, "Init data wrapper failed!") @@ -245,7 +249,6 @@ retry: } else { writeLimit = rate.Limit(config.WriteRate) } - client.readLimiter = rate.NewLimiter(readLimit, defaultReadLimitBurst) client.writeLimiter = rate.NewLimiter(writeLimit, defaultWriteLimitBurst) @@ -274,6 +277,22 @@ func (client *ExtentClient) GetEnablePosixAcl() bool { return client.dataWrapper.EnablePosixAcl } +func (client *ExtentClient) GetFlowInfo() (*proto.ClientReportLimitInfo, bool) { + log.LogInfof("action[ExtentClient.GetFlowInfo]") + return client.LimitManager.GetFlowInfo() +} + +func (client *ExtentClient) UpdateFlowInfo(limit *proto.LimitRsp2Client) { + log.LogInfof("action[UpdateFlowInfo.UpdateFlowInfo]") + client.LimitManager.SetClientLimit(limit) + return +} + +func (client *ExtentClient) SetClientID(id uint64) (err error) { + client.LimitManager.ID = id + return +} + // Open request shall grab the lock until request is sent to the request channel func (client *ExtentClient) OpenStream(inode uint64) error { client.streamerLock.Lock() @@ -619,3 +638,7 @@ func (client *ExtentClient) UpdateDataPartitionForColdVolume() error { func (client *ExtentClient) IsPreloadMode() bool { return client.preload } + +func (client *ExtentClient) UploadFlowInfo(clientInfo wrapper.SimpleClientInfo) error { + return client.dataWrapper.UploadFlowInfo(clientInfo, false) +} diff --git a/sdk/data/stream/stream_reader.go b/sdk/data/stream/stream_reader.go index e8d78ad86..9152b3016 100644 --- a/sdk/data/stream/stream_reader.go +++ b/sdk/data/stream/stream_reader.go @@ -123,7 +123,7 @@ func (s *Streamer) read(data []byte, offset int, size int) (total int, err error ctx := context.Background() s.client.readLimiter.Wait(ctx) - + s.client.LimitManager.ReadAlloc(ctx, size) requests = s.extents.PrepareReadRequests(offset, size, data) for _, req := range requests { if req.ExtentKey == nil { diff --git a/sdk/data/stream/stream_writer.go b/sdk/data/stream/stream_writer.go index 15e35a5f8..ff2947303 100644 --- a/sdk/data/stream/stream_writer.go +++ b/sdk/data/stream/stream_writer.go @@ -290,6 +290,10 @@ func (s *Streamer) write(data []byte, offset, size, flags int) (total int, err e ctx := context.Background() s.client.writeLimiter.Wait(ctx) + if flags&proto.FlagsCache == 0 { + s.client.LimitManager.WriteAlloc(ctx, size) + } + requests := s.extents.PrepareWriteRequests(offset, size, data) log.LogDebugf("Streamer write: ino(%v) prepared requests(%v)", s.inode, requests) diff --git a/sdk/data/wrapper/wrapper.go b/sdk/data/wrapper/wrapper.go index 9d9b5a6fd..6e853175e 100644 --- a/sdk/data/wrapper/wrapper.go +++ b/sdk/data/wrapper/wrapper.go @@ -37,6 +37,12 @@ type DataPartitionView struct { DataPartitions []*DataPartition } +type SimpleClientInfo interface { + GetFlowInfo() (*proto.ClientReportLimitInfo, bool) + UpdateFlowInfo(limit *proto.LimitRsp2Client) + SetClientID(id uint64) error +} + // Wrapper TODO rename. This name does not reflect what it is doing. type Wrapper struct { sync.RWMutex @@ -63,7 +69,7 @@ type Wrapper struct { } // NewDataPartitionWrapper returns a new data partition wrapper. -func NewDataPartitionWrapper(volName string, masters []string, preload bool) (w *Wrapper, err error) { +func NewDataPartitionWrapper(clientInfo SimpleClientInfo, volName string, masters []string, preload bool) (w *Wrapper, err error) { w = new(Wrapper) w.stopC = make(chan struct{}) w.masters = masters @@ -76,10 +82,14 @@ func NewDataPartitionWrapper(volName string, masters []string, preload bool) (w err = errors.Trace(err, "NewDataPartitionWrapper:") return } - if err = w.getSimpleVolView(); err != nil { + + if err = w.GetSimpleVolView(); err != nil { err = errors.Trace(err, "NewDataPartitionWrapper:") return } + + w.UploadFlowInfo(clientInfo, true) + if err = w.initDpSelector(); err != nil { log.LogErrorf("NewDataPartitionWrapper: init initDpSelector failed, [%v]", err) } @@ -90,6 +100,7 @@ func NewDataPartitionWrapper(volName string, masters []string, preload bool) (w if err = w.updateDataNodeStatus(); err != nil { log.LogErrorf("NewDataPartitionWrapper: init DataNodeStatus failed, [%v]", err) } + go w.uploadFlowInfoByTick(clientInfo) go w.update() return } @@ -121,11 +132,11 @@ func (w *Wrapper) updateClusterInfo() (err error) { return } -func (w *Wrapper) getSimpleVolView() (err error) { +func (w *Wrapper) GetSimpleVolView() (err error) { var view *proto.SimpleVolView if view, err = w.mc.AdminAPI().GetVolumeSimpleInfo(w.volName); err != nil { - log.LogWarnf("getSimpleVolView: get volume simple info fail: volume(%v) err(%v)", w.volName, err) + log.LogWarnf("GetSimpleVolView: get volume simple info fail: volume(%v) err(%v)", w.volName, err) return } w.followerRead = view.FollowerRead @@ -134,12 +145,25 @@ func (w *Wrapper) getSimpleVolView() (err error) { w.volType = view.VolType w.EnablePosixAcl = view.EnablePosixAcl - log.LogInfof("getSimpleVolView: get volume simple info: ID(%v) name(%v) owner(%v) status(%v) capacity(%v) "+ + log.LogInfof("GetSimpleVolView: get volume simple info: ID(%v) name(%v) owner(%v) status(%v) capacity(%v) "+ "metaReplicas(%v) dataReplicas(%v) mpCnt(%v) dpCnt(%v) followerRead(%v) createTime(%v) dpSelectorName(%v) "+ - "dpSelectorParm(%v)", + "dpSelectorParm(%v) qoslimit client id(%v)", view.ID, view.Name, view.Owner, view.Status, view.Capacity, view.MpReplicaNum, view.DpReplicaNum, view.MpCnt, view.DpCnt, view.FollowerRead, view.CreateTime, view.DpSelectorName, view.DpSelectorParm) - return nil + + return +} + +func (w *Wrapper) uploadFlowInfoByTick(clientInfo SimpleClientInfo) { + ticker := time.NewTicker(3 * time.Second) + for { + select { + case <-ticker.C: + w.UploadFlowInfo(clientInfo, false) + case <-w.stopC: + return + } + } } func (w *Wrapper) update() { @@ -147,7 +171,7 @@ func (w *Wrapper) update() { for { select { case <-ticker.C: - w.updateSimpleVolView() + w.UpdateSimpleVolView() w.updateDataPartition(false) w.updateDataNodeStatus() case <-w.stopC: @@ -156,7 +180,34 @@ func (w *Wrapper) update() { } } -func (w *Wrapper) updateSimpleVolView() (err error) { +func (w *Wrapper) UploadFlowInfo(clientInfo SimpleClientInfo, init bool) (err error) { + var limitRsp *proto.LimitRsp2Client + log.LogInfof("action[UploadFlowInfo] tick!") + flowInfo, isNeedReport := clientInfo.GetFlowInfo() + if !isNeedReport { + log.LogInfof("action[UploadFlowInfo] no need report!") + return nil + } + + if limitRsp, err = w.mc.AdminAPI().UploadFlowInfo(w.volName, flowInfo); err != nil { + log.LogWarnf("UpdateSimpleVolView: get volume simple info fail: volume(%v) err(%v)", w.volName, err) + return + } + log.LogInfof("action[UploadFlowInfo] init %v get rsp id [%v]", init, limitRsp.ID) + if init { + if limitRsp.ID == 0 { + err = fmt.Errorf("init client get id 0") + log.LogInfof("action[UploadFlowInfo] err %v", err.Error()) + return + } + log.LogInfof("action[UploadFlowInfo] get id %v", limitRsp.ID) + clientInfo.SetClientID(limitRsp.ID) + } + clientInfo.UpdateFlowInfo(limitRsp) + return +} + +func (w *Wrapper) UpdateSimpleVolView() (err error) { var view *proto.SimpleVolView if view, err = w.mc.AdminAPI().GetVolumeSimpleInfo(w.volName); err != nil { log.LogWarnf("updateSimpleVolView: get volume simple info fail: volume(%v) err(%v)", w.volName, err) @@ -164,13 +215,13 @@ func (w *Wrapper) updateSimpleVolView() (err error) { } if w.followerRead != view.FollowerRead && !w.followerReadClientCfg { - log.LogInfof("updateSimpleVolView: update followerRead from old(%v) to new(%v)", + log.LogDebugf("UpdateSimpleVolView: update followerRead from old(%v) to new(%v)", w.followerRead, view.FollowerRead) w.followerRead = view.FollowerRead } if w.dpSelectorName != view.DpSelectorName || w.dpSelectorParm != view.DpSelectorParm { - log.LogInfof("updateSimpleVolView: update dpSelector from old(%v %v) to new(%v %v)", + log.LogDebugf("UpdateSimpleVolView: update dpSelector from old(%v %v) to new(%v %v)", w.dpSelectorName, w.dpSelectorParm, view.DpSelectorName, view.DpSelectorParm) w.Lock() w.dpSelectorName = view.DpSelectorName @@ -181,6 +232,7 @@ func (w *Wrapper) updateSimpleVolView() (err error) { return nil } + func (w *Wrapper) updateDataPartitionByRsp(isInit bool, DataPartitions []*proto.DataPartitionResponse) (err error) { var convert = func(response *proto.DataPartitionResponse) *DataPartition { diff --git a/sdk/graphql/general.go b/sdk/graphql/general.go index 5f76ef10e..b2d104ea1 100644 --- a/sdk/graphql/general.go +++ b/sdk/graphql/general.go @@ -17,8 +17,8 @@ func main() { us := master.UserService{} parseSchema(us.Schema(), proto.AdminUserAPI, "user") - vs := master.VolumeService{} - parseSchema(vs.Schema(), proto.AdminVolumeAPI, "volume") + //vs := master.VolumeService{} + //parseSchema(vs.Schema(), proto.AdminVolumeAPI, "volume") cs := master.ClusterService{} parseSchema(cs.Schema(), proto.AdminClusterAPI, "cluster") diff --git a/sdk/master/api_admin.go b/sdk/master/api_admin.go index db030ea6e..6e7e97f9d 100644 --- a/sdk/master/api_admin.go +++ b/sdk/master/api_admin.go @@ -16,6 +16,8 @@ package master import ( "encoding/json" + "fmt" + "github.com/cubefs/cubefs/util/log" "net/http" "strconv" @@ -360,6 +362,51 @@ func (api *AdminAPI) GetVolumeSimpleInfo(volName string) (vv *proto.SimpleVolVie return } +func (api *AdminAPI) UploadFlowInfo(volName string, + flowInfo *proto.ClientReportLimitInfo) (vv *proto.LimitRsp2Client, err error) { + var request = newAPIRequest(http.MethodGet, proto.QosUpload) + request.addParam("name", volName) + if flowInfo == nil { + return nil, fmt.Errorf("flowinfo is nil") + } + + request.addParam("qosEnable", "true") + var encoded []byte + if encoded, err = json.Marshal(flowInfo); err != nil { + log.LogInfof("action[GetVolumeSimpleInfoWithFlowInfo] flowinfo failed") + return + } + + request.addBody(encoded) + var buf []byte + if buf, err = api.mc.serveRequest(request); err != nil { + return + } + + vv = &proto.LimitRsp2Client{} + if err = json.Unmarshal(buf, &vv); err != nil { + return + } + log.LogInfof("action[UploadFlowInfo] enable %v", vv.Enable) + return +} + +func (api *AdminAPI) GetVolumeSimpleInfoWithFlowInfo(volName string) (vv *proto.SimpleVolView, err error) { + var request = newAPIRequest(http.MethodGet, proto.AdminGetVol) + request.addParam("name", volName) + request.addParam("init", "true") + + var buf []byte + if buf, err = api.mc.serveRequest(request); err != nil { + return + } + vv = &proto.SimpleVolView{} + if err = json.Unmarshal(buf, &vv); err != nil { + return + } + return +} + func (api *AdminAPI) GetClusterInfo() (ci *proto.ClusterInfo, err error) { var request = newAPIRequest(http.MethodGet, proto.AdminGetIP) var buf []byte diff --git a/sdk/meta/api.go b/sdk/meta/api.go index eab5f884a..e5eb40250 100644 --- a/sdk/meta/api.go +++ b/sdk/meta/api.go @@ -578,7 +578,7 @@ func (mw *MetaWrapper) DentryUpdate_ll(parentID uint64, name string, inode uint6 return } -// Used as a callback by stream sdk +// Allocated as a callback by stream sdk func (mw *MetaWrapper) AppendExtentKey(parentInode, inode uint64, ek proto.ExtentKey, discard []proto.ExtentKey) error { mp := mw.getPartitionByInode(inode) if mp == nil { diff --git a/sdk/meta/meta.go b/sdk/meta/meta.go index 276102d25..76a6d6fcc 100644 --- a/sdk/meta/meta.go +++ b/sdk/meta/meta.go @@ -123,11 +123,11 @@ type MetaWrapper struct { closeCh chan struct{} closeOnce sync.Once - // Used to signal the go routines which are waiting for partition view update + // Allocated to signal the go routines which are waiting for partition view update partMutex sync.Mutex partCond *sync.Cond - // Used to trigger and throttle instant partition updates + // Allocated to trigger and throttle instant partition updates forceUpdate chan struct{} forceUpdateLimit *rate.Limiter EnableSummary bool diff --git a/util/btree/btree.go b/util/btree/btree.go index 10898efe1..2c5ebbd61 100644 --- a/util/btree/btree.go +++ b/util/btree/btree.go @@ -586,7 +586,7 @@ func (n *node) iterate(dir direction, start, stop Item, includeStart bool, hit b return hit, true } -// Used for testing/debugging purposes. +// Allocated for testing/debugging purposes. func (n *node) print(w io.Writer, level int) { fmt.Fprintf(w, "%sNODE:%v\n", strings.Repeat(" ", level), n.items) for _, c := range n.children { diff --git a/util/config/config.go b/util/config/config.go index fcfdcf32d..9112e5437 100644 --- a/util/config/config.go +++ b/util/config/config.go @@ -157,6 +157,15 @@ func (c *Config) GetInt64(key string) int64 { return 0 } +// GetBool returns a int64 value for the config key. +func (c *Config) GetInt64WithDefault(key string, defaultVal int64) int64 { + if val := c.GetInt64(key); val == 0 { + return defaultVal + } else { + return val + } +} + // GetSlice returns an array for the config key. func (c *Config) GetSlice(key string) []interface{} { result, present := c.data[key] diff --git a/util/future.go b/util/future.go new file mode 100644 index 000000000..e619ccba2 --- /dev/null +++ b/util/future.go @@ -0,0 +1,56 @@ +package util + +type respErr struct { + errCh chan error +} + +func (e *respErr) init() { + e.errCh = make(chan error, 1) +} + +func (e *respErr) respond(err error) { + e.errCh <- err + close(e.errCh) +} + +func (e *respErr) error() <-chan error { + return e.errCh +} + +// Future the future +type Future struct { + respErr + respCh chan interface{} +} + +func NewFuture() *Future { + f := &Future{ + respCh: make(chan interface{}, 1), + } + f.init() + return f +} + +func (f *Future) Respond(resp interface{}, err error) { + if err == nil { + f.respCh <- resp + close(f.respCh) + } else { + f.respErr.respond(err) + } +} + +// Response wait response +func (f *Future) Response() (resp interface{}, err error) { + select { + case err = <-f.error(): + return + case resp = <-f.respCh: + return + } +} + +// AsyncResponse export channels +func (f *Future) AsyncResponse() (respCh <-chan interface{}, errCh <-chan error) { + return f.respCh, f.errCh +} diff --git a/util/stat/statistic.go b/util/stat/statistic.go index 1f0f8ceb9..f83e1b8ae 100644 --- a/util/stat/statistic.go +++ b/util/stat/statistic.go @@ -248,7 +248,7 @@ func WriteStat() error { log.LogErrorf("Get process memory failed, err: %v", err) fmt.Fprintf(ioStream, "Get Mem Failed.\n") } else { - fmt.Fprintf(ioStream, "Mem Used(kB): VIRT %-10d RES %-10d\n", virt, res) + fmt.Fprintf(ioStream, "Mem Allocated(kB): VIRT %-10d RES %-10d\n", virt, res) } fmt.Fprintf(ioStream, "%-42s|%10s|%8s|%8s|%8s|%8s|%8s|%8s|%8s|\n", diff --git a/util/unit.go b/util/unit.go index abd3ab3b1..6d7afe4c8 100644 --- a/util/unit.go +++ b/util/unit.go @@ -17,6 +17,7 @@ package util import ( "fmt" "regexp" + "strings" ) const ( @@ -65,6 +66,15 @@ func IsIPV4(val interface{}) bool { return isMatch(ip4, val) } +func GetIp(addr string) (ip string) { + var arr []string + if arr = strings.Split(addr, ":"); len(arr) < 2 { + return + } + ip = strings.Trim(arr[0], " ") + return ip +} + func regexpCompile(str string) *regexp.Regexp { return regexp.MustCompile("^" + str + "$") }