feat(flashnode): supplement flashnode info for http interface flashnode/get.#22975706

Signed-off-by: baihailong <baihailong@oppo.com>

flashnode/cachengine/engine.go

@@ -519,3 +519,11 @@ func (c *CacheEngine) GetEvictCount() int {
 	rateStat := c.lruCache.GetRateStat()
 	return int(rateStat.Evicts)
 }
+
+func (c *CacheEngine) GetMaxAlloc() int64 {
+	return c.config.MaxAlloc
+}
+
+func (c *CacheEngine) GetHasAlloc() int64 {
+	return c.lruCache.GetAllocated()
+}

flashnode/cachengine/lru.go

@@ -35,6 +35,7 @@ type LruCache interface {
 	Status() *Status
 	Len() int
 	GetRateStat() RateStat
+	GetAllocated() int64
 }
 
 type Status struct {
@@ -316,3 +317,7 @@ func (c *fCache) Close() error {
 func (c *fCache) GetRateStat() RateStat {
 	return *c.recent
 }
+
+func (c *fCache) GetAllocated() int64 {
+	return c.allocated
+}

flashnode/flashnode_op.go

@@ -16,6 +16,7 @@ package flashnode
 
 import (
 	"context"
+	"encoding/json"
 	"fmt"
 	"net"
 	"sync"
@@ -60,7 +61,22 @@ func (f *FlashNode) handlePacket(conn net.Conn, p *proto.Packet) (err error) {
 }
 
 func (f *FlashNode) opFlashNodeHeartbeat(conn net.Conn, p *proto.Packet) (err error) {
-	p.PacketOkReply()
+	resp := &proto.FlashNodeHeartbeatResponse{}
+	resp.Stat.Total = int64(f.total)
+	resp.Stat.MaxAlloc = f.cacheEngine.GetMaxAlloc()
+	resp.Stat.HasAlloc = f.cacheEngine.GetHasAlloc()
+	resp.Stat.FreeSpace = resp.Stat.Total - resp.Stat.HasAlloc
+	resp.Stat.HitRate = f.cacheEngine.GetHitRate()
+	resp.Stat.Evicts = f.cacheEngine.GetEvictCount()
+	resp.Stat.ReadRps = f.readRps
+
+	reply, err := json.Marshal(resp)
+	if err != nil {
+		p.PacketErrorWithBody(proto.OpErr, []byte(err.Error()))
+		return
+	}
+	p.PacketOkWithBody(reply)
+
 	if err = p.WriteToConn(conn); err != nil {
 		log.LogErrorf("ack master response: %s", err.Error())
 		return err

master/flash_node.go

@@ -50,8 +50,9 @@ type FlashNode struct {
 
 	sync.RWMutex
 	flashNodeValue
-	ReportTime time.Time
-	IsActive   bool
+	HeartBeatStat proto.FlashNodeHeartBeatStat
+	ReportTime    time.Time
+	IsActive      bool
 }
 
 func newFlashNode(addr, zoneName, clusterID, version string, isEnable bool) *FlashNode {
@@ -95,21 +96,30 @@ func (flashNode *FlashNode) isActiveAndEnable() (ok bool) {
 func (flashNode *FlashNode) getFlashNodeViewInfo() (info *proto.FlashNodeViewInfo) {
 	flashNode.RLock()
 	info = &proto.FlashNodeViewInfo{
-		ID:           flashNode.ID,
-		Addr:         flashNode.Addr,
-		ReportTime:   flashNode.ReportTime,
-		IsActive:     flashNode.IsActive,
-		Version:      flashNode.Version,
-		ZoneName:     flashNode.ZoneName,
-		FlashGroupID: flashNode.FlashGroupID,
-		IsEnable:     flashNode.IsEnable,
+		ID:            flashNode.ID,
+		Addr:          flashNode.Addr,
+		ReportTime:    flashNode.ReportTime,
+		IsActive:      flashNode.IsActive,
+		Version:       flashNode.Version,
+		ZoneName:      flashNode.ZoneName,
+		FlashGroupID:  flashNode.FlashGroupID,
+		IsEnable:      flashNode.IsEnable,
+		HeartBeatStat: flashNode.HeartBeatStat,
 	}
 	flashNode.RUnlock()
 	return
 }
 
+func (flashNode *FlashNode) updateFlashNodeStatHeartbeat(stat proto.FlashNodeHeartBeatStat) {
+	log.LogInfof("updateFlashNodeStatHeartbeat, flashNode:%v, heartBeatStat[%v], time:%v", flashNode.Addr, stat, time.Now().Format("2006-01-02 15:04:05"))
+	flashNode.Lock()
+	flashNode.HeartBeatStat = stat
+	flashNode.Unlock()
+}
+
 // TODO: sync with proto.FlashNodeHeartbeatResponse.
 func (c *Cluster) syncFlashNodeHeartbeatTasks(tasks []*proto.AdminTask) {
+	var packet *proto.Packet
 	for _, t := range tasks {
 		if t == nil {
 			continue
@@ -119,11 +129,19 @@ func (c *Cluster) syncFlashNodeHeartbeatTasks(tasks []*proto.AdminTask) {
 			log.LogWarn(fmt.Sprintf("action[syncFlashNodeHeartbeatTasks],nodeAddr:%v,taskID:%v,err:%v", t.OperatorAddr, t.ID, err.Error()))
 			continue
 		}
-		if _, err = node.TaskManager.syncSendAdminTask(t); err != nil {
+		if packet, err = node.TaskManager.syncSendAdminTask(t); err != nil {
 			log.LogError(fmt.Sprintf("action[syncFlashNodeHeartbeatTasks],nodeAddr:%v,taskID:%v,err:%v", t.OperatorAddr, t.ID, err.Error()))
 			continue
 		}
 		node.setActive()
+
+		resp := &proto.FlashNodeHeartbeatResponse{}
+		err = json.Unmarshal(packet.Data, resp)
+		if err != nil {
+			log.LogErrorf("Failed to unmarshal response: %v", err)
+			continue
+		}
+		node.updateFlashNodeStatHeartbeat(resp.Stat)
 	}
 }
 

proto/admin_proto.go

@@ -930,12 +930,23 @@ type LcNodeHeartbeatResponse struct {
 	SnapshotScanningTasks map[string]*SnapshotVerDelTaskResponse
 }
 
+type FlashNodeHeartBeatStat struct {
+	Total     int64
+	MaxAlloc  int64
+	HasAlloc  int64
+	FreeSpace int64
+	HitRate   float64
+	Evicts    int
+	ReadRps   int
+}
+
 // FlashNodeHeartbeatResponse defines the response to the flash node heartbeat.
 type FlashNodeHeartbeatResponse struct {
 	Status   uint8
 	Result   string
 	Version  string
 	ZoneName string
+	Stat     FlashNodeHeartBeatStat
 }
 
 // DeleteFileRequest defines the request to delete a file.

proto/distributed_cache.go

@@ -218,14 +218,15 @@ type FlashGroupAdminView struct {
 }
 
 type FlashNodeViewInfo struct {
-	ID           uint64
-	Addr         string
-	ReportTime   time.Time
-	IsActive     bool
-	Version      string
-	ZoneName     string
-	FlashGroupID uint64
-	IsEnable     bool
+	ID            uint64
+	Addr          string
+	ReportTime    time.Time
+	IsActive      bool
+	Version       string
+	ZoneName      string
+	FlashGroupID  uint64
+	IsEnable      bool
+	HeartBeatStat FlashNodeHeartBeatStat
 }
 
 type FlashNodeStat struct {