Enhancement: add admin nodeInfo http api

Signed-off-by: justice <justice_103@126.com>

cmd/master/scripts/manage_admin_getnodeinfo.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+curl -v "http://192.168.0.11:17010/admin/getNodeInfo"
+

cmd/master/scripts/manage_admin_setnodeinfo.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+#curl -v "http://192.168.0.11:17010/admin/setNodeInfo?batchCount=100"
+curl -v "http://192.168.0.11:17010/admin/setNodeInfo?batchCount=100&markDeleteRate=100"
+

cmd/master/scripts/manage_mn_get_params.sh

@@ -1,3 +0,0 @@
-#!/bin/bash
-#curl -v "http://127.0.0.1/metaNode/getParams?hosts=127.0.0.1:9021,127.0.0.2:9021"  | python -m json.tool
-curl -v "http://127.0.0.1/metaNode/getParams"  | python -m json.tool

cmd/master/scripts/manage_mn_set_params.sh

@@ -1,3 +0,0 @@
-#!/bin/bash
-#curl -v "http://127.0.0.1/metaNode/setParams?batchCount=100"  | python -m json.tool
-curl -v "http://127.0.0.1/metaNode/setParams?batchCount=100&hosts=127.0.0.1:9021,127.0.0.2:9021"  | python -m json.tool

datanode/nodeinfo.go

@@ -0,0 +1,42 @@
+package datanode
+
+import (
+	"time"
+
+	"github.com/chubaofs/chubaofs/util/log"
+)
+
+const (
+	UpdateNodeInfoTicket = 1 * time.Minute
+)
+
+var (
+	nodeInfoStopC = make(chan struct{}, 0)
+)
+
+func (m *DataNode) startUpdateNodeInfo() {
+	ticker := time.NewTicker(UpdateNodeInfoTicket)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-nodeInfoStopC:
+			log.LogInfo("metanode nodeinfo goroutine stopped")
+			return
+		case <-ticker.C:
+			m.updateNodeInfo()
+		}
+	}
+}
+
+func (m *DataNode) stopUpdateNodeInfo() {
+	nodeInfoStopC <- struct{}{}
+}
+
+func (m *DataNode) updateNodeInfo() {
+	clusterInfo, err := MasterClient.AdminAPI().GetClusterInfo()
+	if err != nil {
+		log.LogErrorf("[register] %s", err.Error())
+		return
+	}
+	m.SetMarkDeleteRate(clusterInfo.DataNodeDeleteLimitRate)
+}

datanode/server.go

@@ -15,6 +15,7 @@
 package datanode
 
 import (
+	"context"
 	"fmt"
 	"net"
 	"net/http"
@@ -38,6 +39,7 @@ import (
 	"github.com/chubaofs/chubaofs/util/config"
 	"github.com/chubaofs/chubaofs/util/exporter"
 	"github.com/chubaofs/chubaofs/util/log"
+	"golang.org/x/time/rate"
 )
 
 var (
@@ -50,6 +52,11 @@ var (
 	MasterClient        = masterSDK.NewMasterClient(nil, false)
 )
 
+const (
+	defaultMarkDeleteLimitRate  = rate.Inf
+	defaultMarkDeleteLimitBurst = 128
+)
+
 const (
 	DefaultZoneName         = proto.DefaultZoneName
 	DefaultRaftDir          = "raft"
@@ -92,6 +99,9 @@ type DataNode struct {
 	stopC       chan bool
 
 	control common.Control
+
+	markDeleteLimit *rate.Limiter
+	ctx             context.Context
 }
 
 func NewServer() *DataNode {
@@ -153,6 +163,8 @@ func doStart(server common.Server, cfg *config.Config) (err error) {
 	}
 	go s.registerHandler()
 
+	go s.startUpdateNodeInfo()
+
 	return
 }
 
@@ -162,6 +174,8 @@ func doShutdown(server common.Server) {
 		return
 	}
 	close(s.stopC)
+
+	s.stopUpdateNodeInfo()
 	s.stopTCPService()
 	s.stopRaftServer()
 }
@@ -191,6 +205,12 @@ func (s *DataNode) parseConfig(cfg *config.Config) (err error) {
 	if s.zoneName == "" {
 		s.zoneName = DefaultZoneName
 	}
+
+	markDeleteLimit := rate.Limit(defaultMarkDeleteLimitRate)
+
+	s.ctx = context.Background()
+	s.markDeleteLimit = rate.NewLimiter(markDeleteLimit, defaultMarkDeleteLimitBurst)
+
 	log.LogDebugf("action[parseConfig] load masterAddrs(%v).", MasterClient.Nodes())
 	log.LogDebugf("action[parseConfig] load port(%v).", s.port)
 	log.LogDebugf("action[parseConfig] load zoneName(%v).", s.zoneName)
@@ -411,6 +431,19 @@ func (s *DataNode) incDiskErrCnt(partitionID uint64, err error, flag uint8) {
 	}
 }
 
+func (s *DataNode) SetMarkDeleteRate(v uint64) {
+	if v > 0 {
+		s.markDeleteLimit.SetLimit(rate.Limit(v))
+	} else {
+		s.markDeleteLimit.SetLimit(rate.Inf)
+	}
+}
+
+func (s *DataNode) GetMarkDeleteRate() uint64 {
+	v := s.markDeleteLimit.Limit()
+	return uint64(v)
+}
+
 func IsDiskErr(errMsg string) bool {
 	if strings.Contains(errMsg, syscall.EIO.Error()) || strings.Contains(errMsg, syscall.EROFS.Error()) {
 		return true

datanode/wrap_operator.go

@@ -81,7 +81,7 @@ func (s *DataNode) OperatePacket(p *repl.Packet, c *net.TCPConn) (err error) {
 	case proto.OpMarkDelete:
 		s.handleMarkDeletePacket(p, c)
 	case proto.OpBatchDeleteExtent:
-		s.handleBatchMarkDeletePacket(p,c)
+		s.handleBatchMarkDeletePacket(p, c)
 	case proto.OpRandomWrite, proto.OpSyncRandomWrite:
 		s.handleRandomWritePacket(p)
 	case proto.OpNotifyReplicasToRepair:
@@ -318,6 +318,7 @@ func (s *DataNode) handleMarkDeletePacket(p *repl.Packet, c net.Conn) {
 	var (
 		err error
 	)
+	s.markDeleteLimit.Wait(s.ctx)
 	partition := p.Object.(*DataPartition)
 	if p.ExtentType == proto.TinyExtentType {
 		ext := new(proto.TinyExtentDeleteRecord)
@@ -346,19 +347,20 @@ func (s *DataNode) handleBatchMarkDeletePacket(p *repl.Packet, c net.Conn) {
 	var (
 		err error
 	)
+	s.markDeleteLimit.Wait(s.ctx)
 	partition := p.Object.(*DataPartition)
 	var exts []*proto.ExtentKey
 	err = json.Unmarshal(p.Data, &exts)
-	store:=partition.ExtentStore()
+	store := partition.ExtentStore()
 	if err == nil {
-		for _,ext:=range exts{
-			log.LogInfof(fmt.Sprintf("recive DeleteExtent (%v) from (%v)",ext,c.RemoteAddr().String()))
+		for _, ext := range exts {
+			log.LogInfof(fmt.Sprintf("recive DeleteExtent (%v) from (%v)", ext, c.RemoteAddr().String()))
 			store.MarkDelete(ext.ExtentId, int64(ext.ExtentOffset), int64(ext.Size))
 		}
 	}
 
 	if err != nil {
-		log.LogErrorf(fmt.Sprintf("(%v) error(%v) data (%v)",p.GetUniqueLogId(),err,string(p.Data)))
+		log.LogErrorf(fmt.Sprintf("(%v) error(%v) data (%v)", p.GetUniqueLogId(), err, string(p.Data)))
 		p.PackErrorBody(ActionMarkDelete, err.Error())
 	} else {
 		p.PacketOkReply()

master/api_service.go

@@ -239,7 +239,12 @@ func (m *Server) getCluster(w http.ResponseWriter, r *http.Request) {
 }
 
 func (m *Server) getIPAddr(w http.ResponseWriter, r *http.Request) {
-	cInfo := &proto.ClusterInfo{Cluster: m.cluster.Name, Ip: strings.Split(r.RemoteAddr, ":")[0]}
+	cInfo := &proto.ClusterInfo{
+		Cluster:                  m.cluster.Name,
+		MetaNodeDeleteBatchCount: m.cluster.cfg.MetaNodeDeleteBatchCount,
+		DataNodeDeleteLimitRate:  m.cluster.cfg.DataNodeDeleteLimitRate,
+		Ip:                       strings.Split(r.RemoteAddr, ":")[0],
+	}
 	sendOkReply(w, r, newSuccessHTTPReply(cInfo))
 }
 
@@ -723,51 +728,41 @@ func (m *Server) decommissionDataNode(w http.ResponseWriter, r *http.Request) {
 	sendOkReply(w, r, newSuccessHTTPReply(rstMsg))
 }
 
-// set metanode some interval params
-func (m *Server) setMetaNodeParams(w http.ResponseWriter, r *http.Request) {
+func (m *Server) setNodeInfoHandler(w http.ResponseWriter, r *http.Request) {
 	var (
-		hosts  []string
 		params map[string]interface{}
 		err    error
 	)
-	if hosts, params, err = parseAndExtractSetMetaNodeParams(r); err != nil {
+	if params, err = parseAndExtractSetNodeInfoParams(r); err != nil {
 		sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
 		return
 	}
 
-	resp := make(map[string]interface{})
-	if batchCount, ok := params[metaNodeDeleteBatchCountKey]; ok {
+	if batchCount, ok := params[nodeDeleteBatchCountKey]; ok {
 		if bc, ok := batchCount.(uint64); ok {
-			if err = m.cluster.setMetaNodeParams(hosts, bc); err != nil {
+			if err = m.cluster.setMetaNodeDeleteBatchCount(bc); err != nil {
 				sendErrReply(w, r, newErrHTTPReply(err))
 				return
 			}
-			resp[metaNodeDeleteBatchCountKey] = bc
 		}
 	}
+	if val, ok := params[nodeMarkDeleteRateKey]; ok {
+		if v, ok := val.(uint64); ok {
+			if err = m.cluster.setDataNodeDeleteLimitRate(v); err != nil {
+				sendErrReply(w, r, newErrHTTPReply(err))
+				return
+			}
+		}
+	}
+	sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("set nodeinfo params %v successfully", params)))
 
-	sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("set metanode params %v successfully", resp)))
 }
 
 // get metanode some interval params
-func (m *Server) getMetaNodeParams(w http.ResponseWriter, r *http.Request) {
-	var (
-		hosts []string
-		err   error
-	)
-	if hosts, err = parseAndExtractGetMetaNodeParams(r); err != nil {
-		sendErrReply(w, r, &proto.HTTPReply{Code: proto.ErrCodeParamError, Msg: err.Error()})
-		return
-	}
-
-	resp := make(map[string]interface{})
-	batchCounts := make(map[string]uint64)
-	if batchCounts, err = m.cluster.getMetaNodeParams(hosts); err != nil {
-		sendErrReply(w, r, newErrHTTPReply(err))
-		return
-	}
-
-	resp[metaNodeDeleteBatchCountKey] = batchCounts
+func (m *Server) getNodeInfoHandler(w http.ResponseWriter, r *http.Request) {
+	resp := make(map[string]string)
+	resp[nodeDeleteBatchCountKey] = fmt.Sprintf("%v", m.cluster.cfg.MetaNodeDeleteBatchCount)
+	resp[nodeMarkDeleteRateKey] = fmt.Sprintf("%v", m.cluster.cfg.DataNodeDeleteLimitRate)
 
 	sendOkReply(w, r, newSuccessHTTPReply(fmt.Sprintf("%v", resp)))
 }
@@ -1432,40 +1427,37 @@ func parseAndExtractThreshold(r *http.Request) (threshold float64, err error) {
 	return
 }
 
-func parseAndExtractGetMetaNodeParams(r *http.Request) (hosts []string, err error) {
+func parseAndExtractSetNodeInfoParams(r *http.Request) (params map[string]interface{}, err error) {
 	if err = r.ParseForm(); err != nil {
 		return
 	}
 	var value string
-	if value = r.FormValue(metaNodeHostsKey); value != "" {
-		hosts = strings.Split(value, ",")
-	}
-	return
-}
-
-func parseAndExtractSetMetaNodeParams(r *http.Request) (hosts []string, params map[string]interface{}, err error) {
-	if err = r.ParseForm(); err != nil {
-		return
-	}
-	var value string
-	if value = r.FormValue(metaNodeHostsKey); value != "" {
-		hosts = strings.Split(value, ",")
-	}
 	noParams := true
 	params = make(map[string]interface{})
-	if value = r.FormValue(metaNodeDeleteBatchCountKey); value != "" {
+	if value = r.FormValue(nodeDeleteBatchCountKey); value != "" {
 		noParams = false
 		var batchCount = uint64(0)
 		batchCount, err = strconv.ParseUint(value, 10, 64)
 		if err != nil {
-			err = unmatchedKey(metaNodeDeleteBatchCountKey)
+			err = unmatchedKey(nodeDeleteBatchCountKey)
 			return
 		}
-		params[metaNodeDeleteBatchCountKey] = batchCount
+		params[nodeDeleteBatchCountKey] = batchCount
+
+	}
+	if value = r.FormValue(nodeMarkDeleteRateKey); value != "" {
+		noParams = false
+		var val = uint64(0)
+		val, err = strconv.ParseUint(value, 10, 64)
+		if err != nil {
+			err = unmatchedKey(nodeMarkDeleteRateKey)
+			return
+		}
+		params[nodeMarkDeleteRateKey] = val
 
 	}
 	if noParams {
-		err = keyNotFound(metaNodeDeleteBatchCountKey)
+		err = keyNotFound(nodeDeleteBatchCountKey)
 		return
 	}
 	return

master/cluster.go

@@ -15,7 +15,6 @@
 package master
 
 import (
-	"encoding/binary"
 	"fmt"
 	"sync"
 	"time"
@@ -27,10 +26,6 @@ import (
 	"github.com/chubaofs/chubaofs/util/log"
 )
 
-const (
-	BatchGoroutineSize = 128 //batch sync send goroutine
-)
-
 // Cluster stores all the cluster-level information.
 type Cluster struct {
 	Name                      string
@@ -1615,87 +1610,30 @@ func (c *Cluster) setMetaNodeThreshold(threshold float32) (err error) {
 	return
 }
 
-func (c *Cluster) getMetaNodesByHosts(hosts []string) []*MetaNode {
-	metaNodes := make([]*MetaNode, 0)
-	if hosts != nil && len(hosts) > 0 {
-		for _, h := range hosts {
-			if m, ok := c.metaNodes.Load(h); ok {
-				if meta, ok := m.(*MetaNode); ok {
-					metaNodes = append(metaNodes, meta)
-				}
-			}
-		}
-	} else {
-		c.metaNodes.Range(func(key interface{}, v interface{}) bool {
-			metaNodes = append(metaNodes, v.(*MetaNode))
-			return true
-		})
+func (c *Cluster) setMetaNodeDeleteBatchCount(val uint64) (err error) {
+	oldVal := c.cfg.MetaNodeDeleteBatchCount
+	c.cfg.MetaNodeDeleteBatchCount = val
+	if err = c.syncPutCluster(); err != nil {
+		log.LogErrorf("action[setMetaNodeDeleteBatchCount] err[%v]", err)
+		c.cfg.MetaNodeDeleteBatchCount = oldVal
+		err = proto.ErrPersistenceByRaft
+		return
 	}
-	return metaNodes
-}
-
-//
-func (c *Cluster) setMetaNodeParams(hosts []string, batchCount uint64) (err error) {
-	metaNodes := c.getMetaNodesByHosts(hosts)
-	tasks := make([]*proto.AdminTask, 0)
-	for _, m := range metaNodes {
-		task := m.buildSetParamsTask(batchCount)
-		tasks = append(tasks, task)
-	}
-	c.addMetaNodeTasks(tasks)
-
 	return
 }
 
-func (c *Cluster) getMetaNodeParams(hosts []string) (batchCounts map[string]uint64, err error) {
-	metaNodes := c.getMetaNodesByHosts(hosts)
-
-	batchCounts = make(map[string]uint64)
-	totalSize := len(metaNodes)
-	if totalSize > BatchGoroutineSize {
-		for i := 0; i < totalSize; {
-			leftSize := totalSize - i
-			size := BatchGoroutineSize
-			if leftSize < BatchGoroutineSize {
-				size = leftSize
-			}
-			c.batchSyncSendAdminTask(metaNodes[i:i+size], batchCounts)
-			i += size
-		}
-	} else {
-		c.batchSyncSendAdminTask(metaNodes, batchCounts)
+func (c *Cluster) setDataNodeDeleteLimitRate(val uint64) (err error) {
+	oldVal := c.cfg.DataNodeDeleteLimitRate
+	c.cfg.DataNodeDeleteLimitRate = val
+	if err = c.syncPutCluster(); err != nil {
+		log.LogErrorf("action[setDataNodeDeleteLimitRate] err[%v]", err)
+		c.cfg.DataNodeDeleteLimitRate = oldVal
+		err = proto.ErrPersistenceByRaft
+		return
 	}
-
 	return
 }
 
-func (c *Cluster) batchSyncSendAdminTask(metaNodes []*MetaNode, resp map[string]uint64) {
-	var (
-		mu sync.Mutex
-		wg sync.WaitGroup
-	)
-	for _, m := range metaNodes {
-		wg.Add(1)
-		go func(m *MetaNode) {
-			defer func() {
-				wg.Done()
-			}()
-			task := m.buildGetParamsTask()
-			var packet *proto.Packet
-			var e error
-			if packet, e = m.Sender.syncSendAdminTask(task); e != nil {
-				log.LogErrorf("getMetaNodeParams error: %v", e)
-				return
-			}
-			bc := binary.BigEndian.Uint64(packet.Data)
-			mu.Lock()
-			resp[m.Addr] = bc
-			mu.Unlock()
-		}(m)
-	}
-	wg.Wait()
-}
-
 func (c *Cluster) setDisableAutoAllocate(disableAutoAllocate bool) (err error) {
 	oldFlag := c.DisableAutoAllocate
 	c.DisableAutoAllocate = disableAutoAllocate

master/config.go

@@ -81,6 +81,8 @@ type clusterConfig struct {
 	numberOfDataPartitionsToLoad        int
 	nodeSetCapacity                     int
 	MetaNodeThreshold                   float32
+	MetaNodeDeleteBatchCount            uint64 //metanode delete batch count
+	DataNodeDeleteLimitRate             uint64 //datanode delete limit rate
 	peers                               []raftstore.PeerAddress
 	peerAddrs                           []string
 	heartbeatPort                       int64

master/const.go

@@ -23,32 +23,33 @@ import (
 
 // Keys in the request
 const (
-	addrKey                     = "addr"
-	diskPathKey                 = "disk"
-	nameKey                     = "name"
-	idKey                       = "id"
-	countKey                    = "count"
-	startKey                    = "start"
-	enableKey                   = "enable"
-	thresholdKey                = "threshold"
-	dataPartitionSizeKey        = "size"
-	metaPartitionCountKey       = "mpCount"
-	volCapacityKey              = "capacity"
-	volOwnerKey                 = "owner"
-	volAuthKey                  = "authKey"
-	replicaNumKey               = "replicaNum"
-	followerReadKey             = "followerRead"
-	authenticateKey             = "authenticate"
-	akKey                       = "ak"
-	keywordsKey                 = "keywords"
-	zoneNameKey                 = "zoneName"
-	crossZoneKey                = "crossZone"
-	tokenKey                    = "token"
-	tokenTypeKey                = "tokenType"
-	enableTokenKey              = "enableToken"
-	userKey                     = "user"
-	metaNodeDeleteBatchCountKey = "batchCount"
-	metaNodeHostsKey            = "hosts"
+	addrKey                 = "addr"
+	diskPathKey             = "disk"
+	nameKey                 = "name"
+	idKey                   = "id"
+	countKey                = "count"
+	startKey                = "start"
+	enableKey               = "enable"
+	thresholdKey            = "threshold"
+	dataPartitionSizeKey    = "size"
+	metaPartitionCountKey   = "mpCount"
+	volCapacityKey          = "capacity"
+	volOwnerKey             = "owner"
+	volAuthKey              = "authKey"
+	replicaNumKey           = "replicaNum"
+	followerReadKey         = "followerRead"
+	authenticateKey         = "authenticate"
+	akKey                   = "ak"
+	keywordsKey             = "keywords"
+	zoneNameKey             = "zoneName"
+	crossZoneKey            = "crossZone"
+	tokenKey                = "token"
+	tokenTypeKey            = "tokenType"
+	enableTokenKey          = "enableToken"
+	userKey                 = "user"
+	nodeHostsKey            = "hosts"
+	nodeDeleteBatchCountKey = "batchCount"
+	nodeMarkDeleteRateKey   = "markDeleteRate"
 )
 
 const (

master/data_node.go

@@ -25,14 +25,14 @@ import (
 
 // DataNode stores all the information about a data node
 type DataNode struct {
-	Total                     uint64 `json:"TotalWeight"`
-	Used                      uint64 `json:"UsedWeight"`
-	AvailableSpace            uint64
-	ID                        uint64
-	ZoneName                  string `json:"Zone"`
-	Addr                      string
-	ReportTime                time.Time
-	isActive                  bool
+	Total          uint64 `json:"TotalWeight"`
+	Used           uint64 `json:"UsedWeight"`
+	AvailableSpace uint64
+	ID             uint64
+	ZoneName       string `json:"Zone"`
+	Addr           string
+	ReportTime     time.Time
+	isActive       bool
 	sync.RWMutex
 	UsageRatio                float64 // used / total space
 	SelectedTimes             uint64  // number times that this datanode has been selected as the location for a data partition.

master/http_server.go

@@ -204,11 +204,11 @@ func (m *Server) registerAPIRoutes(router *mux.Router) {
 		Path(proto.DecommissionDisk).
 		HandlerFunc(m.decommissionDisk)
 	router.NewRoute().Methods(http.MethodGet, http.MethodPost).
-		Path(proto.AdminSetMetaNodeParams).
-		HandlerFunc(m.setMetaNodeParams)
+		Path(proto.AdminSetNodeInfo).
+		HandlerFunc(m.setNodeInfoHandler)
 	router.NewRoute().Methods(http.MethodGet, http.MethodPost).
-		Path(proto.AdminGetMetaNodeParams).
-		HandlerFunc(m.getMetaNodeParams)
+		Path(proto.AdminGetNodeInfo).
+		HandlerFunc(m.getNodeInfoHandler)
 
 	// user management APIs
 	router.NewRoute().Methods(http.MethodPost).

master/meta_node.go

@@ -148,17 +148,3 @@ func (metaNode *MetaNode) checkHeartbeat() {
 		metaNode.IsActive = false
 	}
 }
-
-func (metaNode *MetaNode) buildSetParamsTask(batchCount uint64) (task *proto.AdminTask) {
-	request := &proto.SetMetaNodeParamsRequest{
-		BatchCount: batchCount,
-	}
-	task = proto.NewAdminTask(proto.OpSetMetaNodeParams, metaNode.Addr, request)
-	return
-}
-
-func (metaNode *MetaNode) buildGetParamsTask() (task *proto.AdminTask) {
-	request := &proto.GetMetaNodeParamsRequest{}
-	task = proto.NewAdminTask(proto.OpGetMetaNodeParams, metaNode.Addr, request)
-	return
-}

metanode/manager.go

@@ -162,10 +162,6 @@ func (m *metadataManager) HandleMetadataOperation(conn net.Conn, p *Packet,
 		err = m.opAppendMultipart(conn, p, remoteAddr)
 	case proto.OpGetMultipart:
 		err = m.opGetMultipart(conn, p, remoteAddr)
-	case proto.OpSetMetaNodeParams:
-		err = m.opSetMetaNodeParams(conn, p, remoteAddr)
-	case proto.OpGetMetaNodeParams:
-		err = m.opGetMetaNodeParams(conn, p, remoteAddr)
 	default:
 		err = fmt.Errorf("%s unknown Opcode: %d, reqId: %d", remoteAddr,
 			p.Opcode, p.GetReqID())

metanode/manager_op.go

@@ -16,7 +16,6 @@ package metanode
 
 import (
 	"bytes"
-	"encoding/binary"
 	"encoding/json"
 	"fmt"
 	"net"
@@ -98,41 +97,6 @@ end:
 	return
 }
 
-func (m *metadataManager) opSetMetaNodeParams(conn net.Conn, p *Packet, remoteAddr string) (err error) {
-	var (
-		req       = &proto.SetMetaNodeParamsRequest{}
-		adminTask = &proto.AdminTask{
-			Request: req,
-		}
-	)
-	if err = json.Unmarshal(p.Data, adminTask); err != nil {
-		p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
-		m.respondToClient(conn, p)
-		return
-	}
-
-	if req.BatchCount <= 0 {
-		req.BatchCount = uint64(BatchCounts)
-	}
-	if req.BatchCount > 4096 {
-		req.BatchCount = uint64(4096)
-	}
-
-	SetDeleteBatchCount(req.BatchCount)
-
-	return
-}
-
-func (m *metadataManager) opGetMetaNodeParams(conn net.Conn, p *Packet, remoteAddr string) (err error) {
-	bc := DeleteBatchCount()
-	data := make([]byte, 8)
-	binary.BigEndian.PutUint64(data, bc)
-	p.PacketOkWithBody(data)
-	m.respondToClient(conn, p)
-
-	return
-}
-
 func (m *metadataManager) opCreateMetaPartition(conn net.Conn, p *Packet,
 	remoteAddr string) (err error) {
 	defer func() {

metanode/metanode.go

@@ -122,6 +122,8 @@ func doStart(s common.Server, cfg *config.Config) (err error) {
 		return
 	}
 
+	go m.startUpdateNodeInfo()
+
 	exporter.Init(cfg.GetString("role"), cfg)
 
 	// check local partition compare with master ,if lack,then not start
@@ -143,6 +145,7 @@ func doShutdown(s common.Server) {
 	if !ok {
 		return
 	}
+	m.stopUpdateNodeInfo()
 	// shutdown node and release the resource
 	m.stopServer()
 	m.stopMetaManager()
@@ -277,7 +280,7 @@ func (m *MetaNode) register() (err error) {
 	var nodeAddress string
 	for {
 		if step < 1 {
-			clusterInfo, err = getClientIP()
+			clusterInfo, err = getClusterInfo()
 			if err != nil {
 				log.LogErrorf("[register] %s", err.Error())
 				continue
@@ -305,7 +308,7 @@ func NewServer() *MetaNode {
 	return &MetaNode{}
 }
 
-func getClientIP() (ci *proto.ClusterInfo, err error) {
+func getClusterInfo() (ci *proto.ClusterInfo, err error) {
 	ci, err = masterClient.AdminAPI().GetClusterInfo()
 	return
 }

metanode/nodeinfo.go

@@ -0,0 +1,61 @@
+package metanode
+
+import (
+	"sync/atomic"
+	"time"
+
+	"github.com/chubaofs/chubaofs/util/log"
+)
+
+const (
+	UpdateNodeInfoTicket     = 1 * time.Minute
+	DefaultDeleteBatchCounts = 128
+)
+
+type NodeInfo struct {
+	deleteBatchCount uint64
+}
+
+var (
+	nodeInfo      = &NodeInfo{}
+	nodeInfoStopC = make(chan struct{}, 0)
+)
+
+func DeleteBatchCount() uint64 {
+	val := atomic.LoadUint64(&nodeInfo.deleteBatchCount)
+	if val == 0 {
+		val = DefaultDeleteBatchCounts
+	}
+	return val
+}
+
+func SetDeleteBatchCount(val uint64) {
+	atomic.StoreUint64(&nodeInfo.deleteBatchCount, val)
+}
+
+func (m *MetaNode) startUpdateNodeInfo() {
+	ticker := time.NewTicker(UpdateNodeInfoTicket)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-nodeInfoStopC:
+			log.LogInfo("metanode nodeinfo gorutine stopped")
+			return
+		case <-ticker.C:
+			m.updateNodeInfo()
+		}
+	}
+}
+
+func (m *MetaNode) stopUpdateNodeInfo() {
+	nodeInfoStopC <- struct{}{}
+}
+
+func (m *MetaNode) updateNodeInfo() {
+	clusterInfo, err := getClusterInfo()
+	if err != nil {
+		log.LogErrorf("[updateNodeInfo] %s", err.Error())
+		return
+	}
+	SetDeleteBatchCount(clusterInfo.MetaNodeDeleteBatchCount)
+}

metanode/partition_free_list.go

@@ -30,30 +30,13 @@ import (
 const (
 	AsyncDeleteInterval      = 10 * time.Second
 	UpdateVolTicket          = 2 * time.Minute
-	BatchCounts              = 500
+	BatchCounts              = 128
 	OpenRWAppendOpt          = os.O_CREATE | os.O_RDWR | os.O_APPEND
 	TempFileValidTime        = 86400 //units: sec
 	DeleteInodeFileExtension = "INODE_DEL"
 	DeleteWorkerCnt          = 10
 )
 
-var (
-	flMu             sync.RWMutex
-	deleteBatchCount = uint64(BatchCounts)
-)
-
-func DeleteBatchCount() (batchCount uint64) {
-	flMu.RLock()
-	flMu.RUnlock()
-	return deleteBatchCount
-}
-
-func SetDeleteBatchCount(batchCount uint64) {
-	flMu.Lock()
-	deleteBatchCount = batchCount
-	flMu.Unlock()
-}
-
 func (mp *metaPartition) startFreeList() (err error) {
 	if mp.delInodeFp, err = os.OpenFile(path.Join(mp.config.RootDir,
 		DeleteInodeFileExtension), OpenRWAppendOpt, 0644); err != nil {
@@ -156,7 +139,7 @@ func (mp *metaPartition) deleteWorker() {
 // delete Extents by Partition,and find all successDelete inode
 func (mp *metaPartition) batchDeleteExtentsByPartition(partitionDeleteExtents map[uint64][]*proto.ExtentKey, allInodes []*Inode) (shouldCommit []*Inode) {
 	occurErrors := make(map[uint64]error)
-	shouldCommit = make([]*Inode, 0, BatchCounts)
+	shouldCommit = make([]*Inode, 0, DeleteBatchCount())
 	var (
 		wg   sync.WaitGroup
 		lock sync.Mutex
@@ -203,7 +186,7 @@ func (mp *metaPartition) deleteMarkedInodes(inoSlice []uint64) {
 			log.LogErrorf(fmt.Sprintf("metaPartition(%v) deleteMarkedInodes panic (%v)", mp.config.PartitionId, r))
 		}
 	}()
-	shouldCommit := make([]*Inode, 0, BatchCounts)
+	shouldCommit := make([]*Inode, 0, DeleteBatchCount())
 	allDeleteExtents := make(map[string]uint64)
 	deleteExtentsByPartition := make(map[uint64][]*proto.ExtentKey)
 	allInodes := make([]*Inode, 0)

proto/admin_proto.go

@@ -34,8 +34,8 @@ const (
 	AdminCreateMetaPartition       = "/metaPartition/create"
 	AdminSetMetaNodeThreshold      = "/threshold/set"
 	AdminListVols                  = "/vol/list"
-	AdminSetMetaNodeParams         = "/metaNode/setParams"
-	AdminGetMetaNodeParams         = "/metaNode/getParams"
+	AdminSetNodeInfo               = "/admin/setNodeInfo"
+	AdminGetNodeInfo               = "/admin/getNodeInfo"
 
 	// Client APIs
 	ClientDataPartitions = "/client/partitions"
@@ -120,8 +120,10 @@ type RegisterMetaNodeResp struct {
 
 // ClusterInfo defines the cluster infomation.
 type ClusterInfo struct {
-	Cluster string
-	Ip      string
+	Cluster                  string
+	Ip                       string
+	MetaNodeDeleteBatchCount uint64
+	DataNodeDeleteLimitRate  uint64
 }
 
 // CreateDataPartitionRequest defines the request to create a data partition.
@@ -234,20 +236,6 @@ type HeartBeatRequest struct {
 	MasterAddr string
 }
 
-type SetMetaNodeParamsRequest struct {
-	BatchCount uint64
-}
-
-type SetMetaNodeParamsResponse struct {
-}
-
-type GetMetaNodeParamsRequest struct {
-}
-
-type GetMetaNodeParamsResponse struct {
-	BatchCount uint64
-}
-
 // PartitionReport defines the partition report.
 type PartitionReport struct {
 	VolName         string

proto/packet.go

@@ -109,8 +109,6 @@ const (
 	OpAddMetaPartitionRaftMember    uint8 = 0x46
 	OpRemoveMetaPartitionRaftMember uint8 = 0x47
 	OpMetaPartitionTryToLeader      uint8 = 0x48
-	OpSetMetaNodeParams             uint8 = 0x49
-	OpGetMetaNodeParams             uint8 = 0x4A
 
 	// Operations: Master -> DataNode
 	OpCreateDataPartition           uint8 = 0x60
@@ -377,10 +375,6 @@ func (p *Packet) GetOpMsg() (m string) {
 		m = "OpRemoveMultipart"
 	case OpListMultiparts:
 		m = "OpListMultiparts"
-	case OpSetMetaNodeParams:
-		m = "OpSetMetaNodeParams"
-	case OpGetMetaNodeParams:
-		m = "OpGetMetaNodeParams"
 	case OpBatchDeleteExtent:
 		m = "OpBatchDeleteExtent"
 	}